From 5c127e5fe49766e5f8d69b42331ccd472ce27caa Mon Sep 17 00:00:00 2001 From: Sharon Katz <121527325+sharon-fdm@users.noreply.github.com> Date: Fri, 24 Jul 2026 15:46:14 -0400 Subject: [PATCH] Fix software ingestion lock convoys and unbatched deletes (#49894) **Related issue:** Resolves #49805, Resolves #48719 # Checklist for submitter If some of the following don't apply, delete the relevant line. - [x] Changes file added for user-visible changes in `changes/`, `orbit/changes/` or `ee/fleetd-chrome/changes`. See [Changes files](https://github.com/fleetdm/fleet/blob/main/docs/Contributing/guides/committing-changes.md#changes-files) for more information. - [x] Input data is properly validated, `SELECT *` is avoided, SQL injection is prevented (using placeholders for values in statements), JS inline code is prevented especially for url redirects, and untrusted data interpolated into shell scripts/commands is validated against shell metacharacters. - [x] Timeouts are implemented and retries are limited to avoid infinite loops ## Testing - [x] Added/updated automated tests - [x] Where appropriate, [automated tests simulate multiple hosts and test for host isolation](https://github.com/fleetdm/fleet/blob/main/docs/Contributing/reference/patterns-backend.md#unit-testing) (updates to one hosts's records do not affect another) - [x] QA'd all new/changed functionality manually --- ## Context A customer (~2,500 hosts, v4.89.1) had their DB writer slammed with `DELETE FROM host_software_installed_paths` statements carrying 30,000+ IDs each. These never completed, required repeated manual intervention, and the table grew from 14.5M to 14.8M rows in 2 days. This is #49805. While investigating, Victor linked #48719, a related `software_titles` INSERT lock convoy issue seen in load tests. Both are in the same software ingestion code path (`server/datastore/mysql/software.go`), so this PR fixes both. ## Root cause ### #49805: Unbatched DELETEs on `host_software_installed_paths` When a host's software changes, Fleet computes a delta and deletes stale rows from `host_software_installed_paths`. The function `deleteHostSoftwareInstalledPaths()` issued a **single** `DELETE FROM host_software_installed_paths WHERE id IN (?)` with all IDs expanded by `sqlx.In()`. With 30,000+ IDs and 14.8M rows in the table, these massive statements held row locks for minutes, timed out, and never completed. On the next agent check-in, the same (or larger) DELETE was retried, creating a feedback loop where the table grew unboundedly. Notably, the INSERT function for the same table (`insertHostSoftwareInstalledPaths`) already batched at 500 rows. The DELETE simply lacked the same treatment. ### #48719: INSERT IGNORE lock convoys on `software_titles` (related) When a host reports software that Fleet hasn't seen before, `preInsertSoftwareInventory()` runs `INSERT IGNORE INTO software_titles (...)` inside a `withRetryTxx` transaction. For homogeneous fleets (many hosts sharing the same software catalog, typical for imaged corporate Windows machines), hundreds of concurrent goroutines try to INSERT IGNORE the same title rows simultaneously. Even though `INSERT IGNORE` is a no-op when the row already exists, InnoDB still acquires row/gap locks on the unique index for the duration of the enclosing transaction. With many goroutines holding or waiting on the same index locks, the DB enters a "lock convoy" where sessions serialize on locks they don't actually need. In load tests (40 Fleet instances, 100K hosts, 141 identical Windows software items), this produced 690 average active sessions on the writer and 85s fleet-wide p99. The existing read-first check (`getIncomingSoftwareChecksumsToExistingTitles`) prevents the convoy at steady state. But on cold start (empty `software_titles`, e.g. after cleanup purges orphaned titles), the check finds nothing and all goroutines race to INSERT the same titles. ## How I reproduced it Started MySQL via `docker compose up -d mysql_test`, created a git worktree. ### #49805 `TestHostSoftwareInstalledPathsDeleteExplosion`: Created a host with 500 software items and installed paths, then replaced all software with an entirely new set. This triggers `deleteHostSoftwareInstalledPaths()` with all 500 old IDs in a single unbatched DELETE statement. At 500 IDs the local test completes quickly, but the structure confirms the problem: at 30K+ IDs on production Aurora with 14M rows, these never finish. ### #48719 `TestSoftwareTitlesInsertIgnoreLockConvoy`: Created 50 hosts, each reporting 100 identical software items (simulating a homogeneous fleet). Used a barrier to release all 50 goroutines simultaneously, then measured two phases: 1. **Cold start** (empty `software_titles`): All 50 hosts concurrently call `ds.UpdateHostSoftware()`. 2. **Steady state** (titles exist): Same 50 hosts re-ingest. **Before fix:** | Metric | Cold start | Steady state | |--------|-----------|-------------| | Wall time | 3.0s | 38ms | | Avg per-host | 1,981ms | 29ms | | **Convoy factor** | **79x** | | The 79x slowdown confirms the lock convoy. ## How I fixed it ### #49805: Batch the DELETE at 500 Changed `deleteHostSoftwareInstalledPaths()` from a single `DELETE ... WHERE id IN (all IDs)` to a loop that processes 500 IDs per batch, matching the existing INSERT batching pattern in the same file. ### #48719: Three-layer defense against lock convoys **Layer 1 - Move title INSERT IGNORE outside the transaction.** Previously, `INSERT IGNORE INTO software_titles` ran inside `withRetryTxx`, so locks were held for the full transaction duration. Now each title INSERT is executed via `ds.writer(ctx).ExecContext()` outside any transaction, auto-committing independently and holding locks for microseconds. **Layer 2 - singleflight per title key.** Added a `singleflight.Group` on the `Datastore` struct. For each title, only one goroutine actually executes the INSERT; concurrent goroutines wait and share the result. **Layer 3 - In-process cache (`sync.Map`).** After a title is inserted, its key is stored in `knownSoftwareTitleKeys`. Subsequent ingestions check the cache first and skip the INSERT entirely. `CleanupSoftwareTitles` clears the cache when it deletes orphaned titles. The three layers work together: the cache handles the common case (title already known), singleflight handles the cold-start race (only one INSERT per title), and auto-commit ensures even the winning INSERT holds locks for microseconds. ## How I tested that it works ### New reproduction tests - `TestSoftwareTitlesInsertIgnoreLockConvoy`: 50 concurrent hosts, 100 identical software items. Measures cold-start convoy factor and verifies all 100 titles are created. - `TestHostSoftwareInstalledPathsDeleteExplosion`: Full software replacement path with 500 items per host, including concurrent hosts. ### Existing test suite Ran all existing software tests including: - `UpdateHostSoftware`, `UpdateHostSoftwareDeadlock`, `PreInsertSoftwareInventory` - `SoftwareTitleUpgradeCodeDriftMatch`, `UpdateHostSoftwareSameBundleIDDifferentNames` - `CleanupSoftwareTitles` (validates cache invalidation works correctly) - `SaveHost`, `SyncHostsSoftware`, and ~80 other subtests All pass. ### After-fix measurements | Metric | Before fix | After fix | |--------|-----------|-----------| | Cold-start wall (50 hosts) | ~3.0s | ~1.4s | | Cold-start avg per-host | ~1,981ms | ~594ms | | Steady-state wall | ~38ms | ~7ms | | Titles created correctly | 100/100 | 100/100 | The remaining cold-start time is from other pipeline operations (`INSERT IGNORE INTO software`, host_software linking), not from `software_titles`. ## Summary by CodeRabbit - **Performance Improvements** - Improved software inventory ingestion under large, concurrent workloads, including more efficient handling of repeated software-title inserts. - Reduced lock contention when many devices report the same titles at the same time. - Batched deletions of installed software-path records to speed up large updates. - **Bug Fixes** - Ensured deterministic, collation-safe software-title deduplication to prevent incorrect or stale title mapping. - Strengthened orphan cleanup behavior so caches are cleared when orphan titles are removed. - **Tests** - Added stress/regression tests for software-title insert contention and large installed-path delete workloads. --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> --- changes/48719-49805-software-ingestion-perf | 1 + server/datastore/mysql/mysql.go | 47 ++- server/datastore/mysql/software.go | 309 +++++++++++++----- .../mysql/software_lock_convoy_test.go | 291 +++++++++++++++++ server/datastore/mysql/testing_utils_test.go | 3 + 5 files changed, 555 insertions(+), 96 deletions(-) create mode 100644 changes/48719-49805-software-ingestion-perf create mode 100644 server/datastore/mysql/software_lock_convoy_test.go diff --git a/changes/48719-49805-software-ingestion-perf b/changes/48719-49805-software-ingestion-perf new file mode 100644 index 0000000000..f1f729a669 --- /dev/null +++ b/changes/48719-49805-software-ingestion-perf @@ -0,0 +1 @@ +- Improved software ingestion performance at scale: batched `host_software_installed_paths` deletes (previously unbounded single statements) and eliminated `software_titles` INSERT lock convoys during concurrent ingestion by moving title inserts outside the main transaction with singleflight deduplication. diff --git a/server/datastore/mysql/mysql.go b/server/datastore/mysql/mysql.go index 6e20b2dba0..45516c5400 100644 --- a/server/datastore/mysql/mysql.go +++ b/server/datastore/mysql/mysql.go @@ -37,6 +37,7 @@ import ( "github.com/jmoiron/sqlx" "go.opentelemetry.io/otel/attribute" semconv "go.opentelemetry.io/otel/semconv/v1.40.0" + "golang.org/x/sync/singleflight" ) const ( @@ -98,8 +99,31 @@ type Datastore struct { // This key is used to encrypt sensitive data stored in the Fleet DB, for example MDM // certificates and keys. serverPrivateKey string + + // knownSoftwareTitleKeys caches title keys that are known to exist in software_titles. + // This eliminates redundant INSERT IGNORE statements during concurrent software ingestion, + // preventing lock convoys on the unique index when many hosts report the same software catalog. + // The cache evicts an arbitrary half of entries once it reaches a fixed size cap to avoid + // unbounded growth on long-lived servers without forcing a full cold start. + knownSoftwareTitleKeys map[string]struct{} + // knownSoftwareTitleKeysMu serializes cache writes and clears; reads use RLock. + knownSoftwareTitleKeysMu sync.RWMutex + + // titleInsertSF deduplicates concurrent INSERT IGNORE INTO software_titles calls for the + // same title key. Only one goroutine per title actually executes the INSERT; others wait + // and share the result. This prevents lock convoys on cold-start (#48719). + titleInsertSF singleflight.Group } +// maxKnownSoftwareTitleKeys caps the in-process software title cache at roughly 100k entries so +// long-lived servers do not retain every title they have ever seen. +const maxKnownSoftwareTitleKeys = 100_000 + +// evictKnownSoftwareTitleKeys removes half the cache when the cap is hit. Keeping the other half +// preserves most steady-state hits while avoiding a full cold start that would reintroduce a burst +// of INSERT IGNORE statements. +const evictKnownSoftwareTitleKeys = maxKnownSoftwareTitleKeys / 2 + // WithPusher sets an APNs pusher for the datastore, used when activating // next activities that require MDM commands. func (ds *Datastore) WithPusher(p nano_push.Pusher) { @@ -284,17 +308,18 @@ func NewDBConnections(cfg config.MysqlConfig, opts ...DBOption) (*common_mysql.D // Use this when you need to share database connections with other bounded context datastores. func NewDatastore(conns *common_mysql.DBConnections, cfg config.MysqlConfig, c clock.Clock) (*Datastore, error) { ds := &Datastore{ - primary: conns.Primary, - replica: conns.Replica, - logger: conns.Options.Logger, - clock: c, - config: cfg, - readReplicaConfig: conns.Options.ReplicaConfig, - writeCh: make(chan itemToWrite), - stmtCache: make(map[string]*sqlx.Stmt), - minLastOpenedAtDiff: conns.Options.MinLastOpenedAtDiff, - serverPrivateKey: conns.Options.PrivateKey, - Datastore: NewAndroidDatastore(conns.Options.Logger, conns.Primary, conns.Replica), + primary: conns.Primary, + replica: conns.Replica, + logger: conns.Options.Logger, + clock: c, + config: cfg, + readReplicaConfig: conns.Options.ReplicaConfig, + writeCh: make(chan itemToWrite), + stmtCache: make(map[string]*sqlx.Stmt), + minLastOpenedAtDiff: conns.Options.MinLastOpenedAtDiff, + serverPrivateKey: conns.Options.PrivateKey, + knownSoftwareTitleKeys: make(map[string]struct{}), + Datastore: NewAndroidDatastore(conns.Options.Logger, conns.Primary, conns.Replica), } go ds.writeChanLoop() diff --git a/server/datastore/mysql/software.go b/server/datastore/mysql/software.go index 8eb6a50d4d..04ae25f20d 100644 --- a/server/datastore/mysql/software.go +++ b/server/datastore/mysql/software.go @@ -72,6 +72,18 @@ var cleanupBatchSize = 1000 // Any remaining orphans will be processed on the next hourly cron cycle. var cleanupMaxIterations = 100 +// softwareTitleCacheKey builds a string key for the in-process cache of known software titles. +// It mirrors the titleKey struct used inside preInsertSoftwareInventory. +func softwareTitleCacheKey(name, source, extensionFor, bundleID string, isKernel bool) string { + return strings.Join([]string{ + strings.ToLower(normalizeForCollation(name)), + source, + extensionFor, + strings.ToLower(bundleID), + strconv.FormatBool(isKernel), + }, fleet.SoftwareFieldSeparator) +} + func softwareSliceToMap(softwareItems []fleet.Software) map[string]fleet.Software { result := make(map[string]fleet.Software, len(softwareItems)) for _, s := range softwareItems { @@ -80,6 +92,52 @@ func softwareSliceToMap(softwareItems []fleet.Software) map[string]fleet.Softwar return result } +func (ds *Datastore) cacheKnownSoftwareTitleKey(key string) { + ds.knownSoftwareTitleKeysMu.Lock() + defer ds.knownSoftwareTitleKeysMu.Unlock() + if _, loaded := ds.knownSoftwareTitleKeys[key]; loaded { + return + } + if len(ds.knownSoftwareTitleKeys) >= maxKnownSoftwareTitleKeys { + ds.evictKnownSoftwareTitleKeysLocked() + } + // Store after potential eviction so the caller's key survives. + ds.knownSoftwareTitleKeys[key] = struct{}{} +} + +func (ds *Datastore) clearKnownSoftwareTitleKeys() { + ds.knownSoftwareTitleKeysMu.Lock() + defer ds.knownSoftwareTitleKeysMu.Unlock() + ds.knownSoftwareTitleKeys = make(map[string]struct{}) +} + +func (ds *Datastore) deleteKnownSoftwareTitleKey(key string) { + ds.knownSoftwareTitleKeysMu.Lock() + defer ds.knownSoftwareTitleKeysMu.Unlock() + delete(ds.knownSoftwareTitleKeys, key) +} + +func (ds *Datastore) hasKnownSoftwareTitleKey(key string) bool { + ds.knownSoftwareTitleKeysMu.RLock() + defer ds.knownSoftwareTitleKeysMu.RUnlock() + _, ok := ds.knownSoftwareTitleKeys[key] + return ok +} + +func (ds *Datastore) evictKnownSoftwareTitleKeysLocked() { + evicted := 0 + // Go map iteration order is randomized, so this evicts an arbitrary half of the cache. + // That is sufficient here because any retained title key still avoids an INSERT IGNORE, and + // arbitrary bulk eviction is much cheaper than maintaining a strict LRU in this hot path. + for key := range ds.knownSoftwareTitleKeys { + delete(ds.knownSoftwareTitleKeys, key) + evicted++ + if evicted >= evictKnownSoftwareTitleKeys { + return + } + } +} + func (ds *Datastore) UpdateHostSoftware(ctx context.Context, hostID uint, software []fleet.Software) (*fleet.UpdateHostSoftwareDBResult, error) { // OTEL instrumentation. It has no-op behavior when OTEL is not enabled. ctx, span := tracer.Start(ctx, "mysql.UpdateHostSoftware", @@ -293,13 +351,19 @@ func deleteHostSoftwareInstalledPaths( return nil } - stmt := `DELETE FROM host_software_installed_paths WHERE id IN (?)` - stmt, args, err := sqlx.In(stmt, toDelete) - if err != nil { - return ctxerr.Wrap(ctx, err, "building delete statement for delete host_software_installed_paths") - } - if _, err := tx.ExecContext(ctx, stmt, args...); err != nil { - return ctxerr.Wrap(ctx, err, "executing delete statement for delete host_software_installed_paths") + const batchSize = 500 + for i := 0; i < len(toDelete); i += batchSize { + end := min(i+batchSize, len(toDelete)) + batch := toDelete[i:end] + + stmt := `DELETE FROM host_software_installed_paths WHERE id IN (?)` + stmt, args, err := sqlx.In(stmt, batch) + if err != nil { + return ctxerr.Wrap(ctx, err, "building delete statement for delete host_software_installed_paths") + } + if _, err := tx.ExecContext(ctx, stmt, args...); err != nil { + return ctxerr.Wrap(ctx, err, "executing delete statement for delete host_software_installed_paths") + } } return nil @@ -1018,51 +1082,111 @@ func (ds *Datastore) preInsertSoftwareInventory( batchSoftware[key] = needsInsert[key] } - // Each batch in its own transaction - return ds.withRetryTxx(ctx, func(tx sqlx.ExtContext) error { - // First insert any needed software titles - newTitlesNeeded := make(map[string]fleet.SoftwareTitle) - for checksum, sw := range batchSoftware { - if _, ok := incomingChecksumsToExistingTitleSummaries[checksum]; !ok { - // there is not an existing software title corresponding to this incoming software version - newTitleName := sw.Name - if sw.BundleIdentifier != "" { - // First check if there's an FMA with this bundle identifier - use its canonical name - if fmaName, ok := fmaNames[sw.BundleIdentifier]; ok { - newTitleName = fmaName - } else { - // Fall back to computed best name from osquery reports - key := titleKey{ - bundleID: sw.BundleIdentifier, - source: sw.Source, - extensionFor: sw.ExtensionFor, - } - if computedName, exists := bestTitleNames[key]; exists { - newTitleName = computedName - } + // Compute which software titles need to be created. + // This is done outside the transaction because the computation is pure (no DB access). + newTitlesNeeded := make(map[string]fleet.SoftwareTitle) + for checksum, sw := range batchSoftware { + if _, ok := incomingChecksumsToExistingTitleSummaries[checksum]; !ok { + // there is not an existing software title corresponding to this incoming software version + newTitleName := sw.Name + if sw.BundleIdentifier != "" { + // First check if there's an FMA with this bundle identifier - use its canonical name + if fmaName, ok := fmaNames[sw.BundleIdentifier]; ok { + newTitleName = fmaName + } else { + // Fall back to computed best name from osquery reports + key := titleKey{ + bundleID: sw.BundleIdentifier, + source: sw.Source, + extensionFor: sw.ExtensionFor, + } + if computedName, exists := bestTitleNames[key]; exists { + newTitleName = computedName } } + } - newTitle := fleet.SoftwareTitle{ - Name: newTitleName, - Source: sw.Source, - ExtensionFor: sw.ExtensionFor, - IsKernel: sw.IsKernel, - } - if sw.BundleIdentifier != "" { - newTitle.BundleIdentifier = ptr.String(sw.BundleIdentifier) - } - if sw.ApplicationID != nil && *sw.ApplicationID != "" { - newTitle.ApplicationID = sw.ApplicationID - } - if sw.UpgradeCode != nil { - // intentionally write both empty and non-empty strings as upgrade codes - newTitle.UpgradeCode = sw.UpgradeCode - } - newTitlesNeeded[checksum] = newTitle + newTitle := fleet.SoftwareTitle{ + Name: newTitleName, + Source: sw.Source, + ExtensionFor: sw.ExtensionFor, + IsKernel: sw.IsKernel, + } + if sw.BundleIdentifier != "" { + newTitle.BundleIdentifier = new(sw.BundleIdentifier) + } + if sw.ApplicationID != nil && *sw.ApplicationID != "" { + newTitle.ApplicationID = sw.ApplicationID + } + if sw.UpgradeCode != nil { + // intentionally write both empty and non-empty strings as upgrade codes + newTitle.UpgradeCode = sw.UpgradeCode + } + newTitlesNeeded[checksum] = newTitle + } + } + + // INSERT IGNORE new software titles OUTSIDE the main transaction (#48719). + // Each INSERT IGNORE is auto-committed independently, so it holds row/gap locks + // for only microseconds instead of the entire transaction duration. This eliminates + // lock convoys when many hosts concurrently report the same software catalog. + if len(newTitlesNeeded) > 0 { + // Build the full set of unique titles (for ID resolution later). + uniqueTitles := make(map[titleKey]fleet.SoftwareTitle) + for _, title := range newTitlesNeeded { + bundleID := "" + if title.BundleIdentifier != nil { + bundleID = *title.BundleIdentifier + } + key := titleKey{ + name: strings.ToLower(normalizeForCollation(title.Name)), + source: title.Source, + extensionFor: title.ExtensionFor, + bundleID: bundleID, + isKernel: title.IsKernel, + } + if _, exists := uniqueTitles[key]; !exists { + uniqueTitles[key] = title } } + // INSERT IGNORE each title individually using auto-commit (outside any transaction). + // singleflight ensures that for each title key, only one goroutine actually + // executes the INSERT; concurrent goroutines wait and share the result. + // The in-process cache prevents future DB hits entirely. + const insertTitleStmt = `INSERT IGNORE INTO software_titles (name, source, extension_for, bundle_identifier, is_kernel, application_id, upgrade_code) VALUES (?,?,?,?,?,?,?)` + for key, title := range uniqueTitles { + cacheKey := softwareTitleCacheKey(title.Name, title.Source, title.ExtensionFor, key.bundleID, title.IsKernel) + if ds.hasKnownSoftwareTitleKey(cacheKey) { + continue + } + // Capture loop variables for the closure. + titleCopy := title + _, sfErr, _ := ds.titleInsertSF.Do(cacheKey, func() (any, error) { + // Double-check cache after winning the singleflight race. + if ds.hasKnownSoftwareTitleKey(cacheKey) { + return nil, nil + } + // Use context.WithoutCancel so the INSERT completes even if the + // leader goroutine's request is canceled mid-flight (#48719). + insertCtx := context.WithoutCancel(ctx) + if _, err := ds.writer(insertCtx).ExecContext(insertCtx, insertTitleStmt, + titleCopy.Name, titleCopy.Source, titleCopy.ExtensionFor, titleCopy.BundleIdentifier, + titleCopy.IsKernel, titleCopy.ApplicationID, titleCopy.UpgradeCode, + ); err != nil { + return nil, ctxerr.Wrap(ctx, err, "pre-insert software_titles") + } + ds.cacheKnownSoftwareTitleKey(cacheKey) + return nil, nil + }) + if sfErr != nil { + return sfErr + } + } + } + + // Each batch in its own transaction (for SELECT title IDs + INSERT software). + return ds.withRetryTxx(ctx, func(tx sqlx.ExtContext) error { // Map to store title IDs for all titles (both existing and new) titleIDsByChecksum := make(map[string]uint, len(incomingChecksumsToExistingTitleSummaries)) @@ -1071,47 +1195,33 @@ func (ds *Datastore) preInsertSoftwareInventory( titleIDsByChecksum[checksum] = titleSummary.ID } if len(newTitlesNeeded) > 0 { - uniqueTitlesToInsert := make(map[titleKey]fleet.SoftwareTitle) + // Build the set of unique titles for the SELECT query. + uniqueTitles := make(map[titleKey]fleet.SoftwareTitle) for _, title := range newTitlesNeeded { bundleID := "" if title.BundleIdentifier != nil { bundleID = *title.BundleIdentifier } key := titleKey{ - // adjust for matching MySQL collation name: strings.ToLower(normalizeForCollation(title.Name)), source: title.Source, extensionFor: title.ExtensionFor, bundleID: bundleID, isKernel: title.IsKernel, } - - if _, exists := uniqueTitlesToInsert[key]; !exists { - uniqueTitlesToInsert[key] = title + if _, exists := uniqueTitles[key]; !exists { + uniqueTitles[key] = title } } - // Insert software titles - const numberOfArgsPerSoftwareTitles = 7 - titlesValues := strings.TrimSuffix(strings.Repeat("(?,?,?,?,?,?,?),", len(uniqueTitlesToInsert)), ",") - titlesStmt := fmt.Sprintf("INSERT IGNORE INTO software_titles (name, source, extension_for, bundle_identifier, is_kernel, application_id, upgrade_code) VALUES %s", titlesValues) - titlesArgs := make([]any, 0, len(uniqueTitlesToInsert)*numberOfArgsPerSoftwareTitles) - - for _, title := range uniqueTitlesToInsert { - titlesArgs = append(titlesArgs, title.Name, title.Source, title.ExtensionFor, title.BundleIdentifier, title.IsKernel, title.ApplicationID, title.UpgradeCode) - } - - if _, err := tx.ExecContext(ctx, titlesStmt, titlesArgs...); err != nil { - return ctxerr.Wrap(ctx, err, "pre-insert software_titles") - } - - // Retrieve the IDs for the titles we just inserted (or that already existed) + // Retrieve the IDs for the titles we just inserted (or that already existed). + // Use uniqueTitles (all unique titles) so we resolve IDs for cached titles too. var retrievedTitleSummaries []fleet.SoftwareTitleSummary - titlePlaceholders := strings.TrimSuffix(strings.Repeat("(?,?,?,?),", len(uniqueTitlesToInsert)), ",") - queryArgs := make([]interface{}, 0, len(uniqueTitlesToInsert)*4) + titlePlaceholders := strings.TrimSuffix(strings.Repeat("(?,?,?,?),", len(uniqueTitles)), ",") + queryArgs := make([]any, 0, len(uniqueTitles)*4) var upgradeCodes []string - for tk := range uniqueTitlesToInsert { - title := uniqueTitlesToInsert[tk] + for tk := range uniqueTitles { + title := uniqueTitles[tk] bundleID := "" if title.BundleIdentifier != nil { bundleID = *title.BundleIdentifier @@ -1239,6 +1349,7 @@ func (ds *Datastore) preInsertSoftwareInventory( } } } + } // Insert software entries @@ -1266,7 +1377,7 @@ func (ds *Datastore) preInsertSoftwareInventory( ) args := make([]any, 0, len(batchKeys)*numberOfArgsPerSoftware) - var missingSoftwareTitles []string + var missingChecksums []string for _, checksum := range batchKeys { sw := batchSoftware[checksum] var titleID *uint @@ -1274,9 +1385,9 @@ func (ds *Datastore) preInsertSoftwareInventory( if id, ok := titleIDsByChecksum[checksum]; ok { titleID = &id } else { - // Track software missing title IDs for debugging - missingSoftwareTitles = append(missingSoftwareTitles, - fmt.Sprintf("%s %s %s", sw.Name, sw.Version, sw.Source)) + // Track software missing title IDs; titles inserted outside the + // transaction may have been deleted by a concurrent CleanupSoftwareTitles. + missingChecksums = append(missingChecksums, checksum) } // Use FMA canonical name if available, otherwise use osquery-reported name. @@ -1310,17 +1421,38 @@ func (ds *Datastore) preInsertSoftwareInventory( ) } - // Log an error if we have software without title IDs - // This shouldn't happen in normal operation. And this code is here to catch bugs. - if len(missingSoftwareTitles) > 0 && ds.logger != nil { - exampleCount := 3 - if len(missingSoftwareTitles) < exampleCount { - exampleCount = len(missingSoftwareTitles) + // When title IDs are missing, a concurrent CleanupSoftwareTitles likely + // deleted the titles we just inserted (they were orphaned briefly outside the + // transaction). Clear those cache entries so they are re-inserted on the next + // agent check-in. The software row proceeds with NULL title_id; the next + // ingestion cycle will re-create the title and link it. + if len(missingChecksums) > 0 { + var examples []string + for _, checksum := range missingChecksums { + sw := batchSoftware[checksum] + if len(examples) < 3 { + examples = append(examples, fmt.Sprintf("%s %s %s", sw.Name, sw.Version, sw.Source)) + } + // Evict from the in-process cache so the next ingestion cycle re-inserts the title. + if title, ok := newTitlesNeeded[checksum]; ok { + bundleID := "" + if title.BundleIdentifier != nil { + bundleID = *title.BundleIdentifier + } + cacheKey := softwareTitleCacheKey(title.Name, title.Source, title.ExtensionFor, bundleID, title.IsKernel) + ds.deleteKnownSoftwareTitleKey(cacheKey) + } + } + // Log rather than return a hard error: the title INSERT is outside the + // transaction, so withRetryTxx cannot re-insert the title on retry. + // The software row proceeds with NULL title_id and the evicted cache + // entry ensures the title is re-created on the next ingestion cycle. + if ds.logger != nil { + ds.logger.ErrorContext(ctx, "inserting software without title_id", + "count", len(missingChecksums), + "examples", strings.Join(examples, "; "), + ) } - ds.logger.ErrorContext(ctx, "inserting software without title_id", - "count", len(missingSoftwareTitles), - "examples", strings.Join(missingSoftwareTitles[:exampleCount], "; "), - ) } if _, err := tx.ExecContext(ctx, stmt, args...); err != nil { @@ -3040,7 +3172,7 @@ func (ds *Datastore) CleanupSoftwareTitles(ctx context.Context) error { return ctxerr.Wrap(ctx, err, "find orphaned software titles for cleanup") } if len(ids) == 0 { - return nil + break } lastID = ids[len(ids)-1] @@ -3055,6 +3187,13 @@ func (ds *Datastore) CleanupSoftwareTitles(ctx context.Context) error { ra, _ := res.RowsAffected() n += ra } + + // If any titles were deleted, clear the in-process title cache so that future + // software ingestions re-insert titles instead of skipping them. + if n > 0 { + ds.clearKnownSoftwareTitleKeys() + } + return nil } diff --git a/server/datastore/mysql/software_lock_convoy_test.go b/server/datastore/mysql/software_lock_convoy_test.go new file mode 100644 index 0000000000..3bf35d3d22 --- /dev/null +++ b/server/datastore/mysql/software_lock_convoy_test.go @@ -0,0 +1,291 @@ +package mysql + +import ( + "context" + "fmt" + "slices" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/fleetdm/fleet/v4/server/fleet" + "github.com/fleetdm/fleet/v4/server/test" + "github.com/jmoiron/sqlx" + "github.com/stretchr/testify/require" + "golang.org/x/sync/errgroup" +) + +// TestSoftwareTitlesInsertIgnoreLockConvoy reproduces the lock convoy described in #48719. +// +// Setup: N hosts all report the SAME software catalog (homogeneous fleet, like imaged +// corporate Windows machines). The software_titles table starts empty (cold start). +// All hosts race to INSERT IGNORE the same titles concurrently. +// +// Expected: With the current code, concurrent INSERT IGNORE statements on the same +// unique-index rows serialize and cause high contention. This test measures timing +// to confirm the convoy is observable even at modest concurrency. +func TestSoftwareTitlesInsertIgnoreLockConvoy(t *testing.T) { + ds := CreateMySQLDS(t) + ctx := context.Background() + + const ( + hostCount = 50 // concurrent hosts + softwareCount = 100 // software items per host (all identical across hosts) + ) + + // Create hosts + hosts := make([]*fleet.Host, hostCount) + for i := range hostCount { + h, err := ds.NewHost(ctx, &fleet.Host{ + OsqueryHostID: new(fmt.Sprintf("convoy-host-%d", i)), + NodeKey: new(fmt.Sprintf("convoy-key-%d", i)), + Platform: "windows", + Hostname: fmt.Sprintf("convoy-host-%d", i), + DetailUpdatedAt: time.Now(), + LabelUpdatedAt: time.Now(), + PolicyUpdatedAt: time.Now(), + SeenTime: time.Now(), + }) + require.NoError(t, err) + hosts[i] = h + } + + // Build a SINGLE software catalog shared by ALL hosts (homogeneous fleet). + // This is the key condition for the lock convoy: every host tries to INSERT IGNORE + // the same software_titles rows. + sharedSoftware := make([]fleet.Software, softwareCount) + for i := range softwareCount { + sharedSoftware[i] = fleet.Software{ + Name: fmt.Sprintf("ConvoyApp %d", i), + Version: "1.0.0", + Source: "programs", + } + } + + // --- Cold-start convoy: all hosts ingest simultaneously with empty software_titles --- + t.Log("Starting cold-start convoy test...") + t.Logf(" Hosts: %d, Software items per host: %d", hostCount, softwareCount) + + var ( + g errgroup.Group + maxElapsed atomic.Int64 + totalMs atomic.Int64 + ready = make(chan struct{}) // barrier to synchronize start + ) + + for i := range hostCount { + hostID := hosts[i].ID + // Copy the slice to avoid data races (UpdateHostSoftware may mutate it in-place). + sw := slices.Clone(sharedSoftware) + g.Go(func() error { + <-ready // wait for all goroutines to be ready + start := time.Now() + _, err := ds.UpdateHostSoftware(ctx, hostID, sw) + elapsed := time.Since(start) + ms := elapsed.Milliseconds() + totalMs.Add(ms) + for { + old := maxElapsed.Load() + if ms <= old || maxElapsed.CompareAndSwap(old, ms) { + break + } + } + if err != nil { + return fmt.Errorf("host %d: %w", hostID, err) + } + return nil + }) + } + + start := time.Now() + close(ready) // release all goroutines at once + err := g.Wait() + wallTime := time.Since(start) + + require.NoError(t, err) + + t.Logf(" Cold-start results:") + t.Logf(" Wall time: %s", wallTime) + t.Logf(" Max single-host ingestion: %dms", maxElapsed.Load()) + t.Logf(" Avg per-host ingestion: %dms", totalMs.Load()/int64(hostCount)) + + // Verify all titles were created + var titleCount int + ExecAdhocSQL(t, ds, func(q sqlx.ExtContext) error { + return sqlx.GetContext(ctx, q, &titleCount, `SELECT COUNT(*) FROM software_titles WHERE source = 'programs'`) + }) + t.Logf(" Software titles created: %d (expected %d)", titleCount, softwareCount) + require.Equal(t, softwareCount, titleCount) + + // --- Steady-state: re-ingest same software (should be fast, read-only path) --- + t.Log("Starting steady-state re-ingestion test...") + maxElapsed.Store(0) + totalMs.Store(0) + ready2 := make(chan struct{}) + + var g2 errgroup.Group + for i := range hostCount { + hostID := hosts[i].ID + sw := slices.Clone(sharedSoftware) + g2.Go(func() error { + <-ready2 + start := time.Now() + _, err := ds.UpdateHostSoftware(ctx, hostID, sw) + elapsed := time.Since(start) + ms := elapsed.Milliseconds() + totalMs.Add(ms) + for { + old := maxElapsed.Load() + if ms <= old || maxElapsed.CompareAndSwap(old, ms) { + break + } + } + if err != nil { + return fmt.Errorf("host %d: %w", hostID, err) + } + return nil + }) + } + + start2 := time.Now() + close(ready2) + err = g2.Wait() + wallTime2 := time.Since(start2) + + require.NoError(t, err) + + t.Logf(" Steady-state results:") + t.Logf(" Wall time: %s", wallTime2) + t.Logf(" Max single-host ingestion: %dms", maxElapsed.Load()) + t.Logf(" Avg per-host ingestion: %dms", totalMs.Load()/int64(hostCount)) + + // The cold-start should be significantly slower than steady-state due to lock contention + if wallTime2 > 0 { + t.Logf("\n Convoy factor (cold wall / steady wall): %.1fx", float64(wallTime)/float64(wallTime2)) + } +} + +// TestHostSoftwareInstalledPathsDeleteExplosion reproduces #49805. +// +// A host with many installed paths gets re-enrolled or its software changes significantly, +// triggering a DELETE FROM host_software_installed_paths WHERE id IN (thousands of IDs) +// in a single unbatched statement. +func TestHostSoftwareInstalledPathsDeleteExplosion(t *testing.T) { + ds := CreateMySQLDS(t) + ctx := context.Background() + + // Create a host + host := test.NewHost(t, ds, "delete-explosion-host", "", "de-key", "de-uuid", time.Now()) + + // Insert a large number of software items, each with an installed path + const softwareCount = 500 // a more modest number than 30k for local testing + software := make([]fleet.Software, softwareCount) + for i := range softwareCount { + software[i] = fleet.Software{ + Name: fmt.Sprintf("DeleteTestApp %d", i), + Version: "1.0.0", + Source: "apps", + } + } + + // First ingestion: establish software + _, err := ds.UpdateHostSoftware(ctx, host.ID, software) + require.NoError(t, err) + + // Get the software IDs that were created + var swIDs []struct { + ID uint `db:"id"` + Name string `db:"name"` + } + ExecAdhocSQL(t, ds, func(q sqlx.ExtContext) error { + return sqlx.SelectContext(ctx, q, &swIDs, + `SELECT id, name FROM software WHERE name LIKE 'DeleteTestApp%' AND source = 'apps'`) + }) + t.Logf("Created %d software entries", len(swIDs)) + + // Directly insert installed paths to build up the table + ExecAdhocSQL(t, ds, func(q sqlx.ExtContext) error { + for _, sw := range swIDs { + _, err := q.ExecContext(ctx, + `INSERT INTO host_software_installed_paths (host_id, software_id, installed_path) VALUES (?, ?, ?)`, + host.ID, sw.ID, fmt.Sprintf("/Applications/%s.app", sw.Name)) + if err != nil { + return err + } + } + return nil + }) + + // Verify the paths are there + var pathCount int + ExecAdhocSQL(t, ds, func(q sqlx.ExtContext) error { + return sqlx.GetContext(ctx, q, &pathCount, + `SELECT COUNT(*) FROM host_software_installed_paths WHERE host_id = ?`, host.ID) + }) + t.Logf("Installed paths for host: %d", pathCount) + require.Equal(t, len(swIDs), pathCount) + + // Now simulate a "full replacement" by reporting all-new software with no overlap. + // This causes ALL existing paths to be deleted in one shot. + newSoftware := make([]fleet.Software, softwareCount) + for i := range softwareCount { + newSoftware[i] = fleet.Software{ + Name: fmt.Sprintf("ReplacementApp %d", i), + Version: "2.0.0", + Source: "apps", + } + } + + // This should trigger a massive DELETE of all old installed paths + startDel := time.Now() + _, err = ds.UpdateHostSoftware(ctx, host.ID, newSoftware) + elapsed := time.Since(startDel) + require.NoError(t, err) + t.Logf("Full software replacement took: %s", elapsed) + + // Now test with concurrent hosts doing the same thing + t.Log("Testing concurrent large deletes...") + const concurrentHosts = 10 + var wg sync.WaitGroup + wg.Add(concurrentHosts) + + // Create hosts and insert paths synchronously to avoid require.*/ExecAdhocSQL panics from goroutines. + concurrentTestHosts := make([]*fleet.Host, concurrentHosts) + for i := range concurrentHosts { + concurrentTestHosts[i] = test.NewHost(t, ds, fmt.Sprintf("concurrent-del-%d", i), "", fmt.Sprintf("cd-key-%d", i), fmt.Sprintf("cd-uuid-%d", i), time.Now()) + + swCopy := slices.Clone(software) + _, err := ds.UpdateHostSoftware(ctx, concurrentTestHosts[i].ID, swCopy) + require.NoError(t, err) + + ExecAdhocSQL(t, ds, func(q sqlx.ExtContext) error { + for _, sw := range swIDs { + _, err := q.ExecContext(ctx, + `INSERT IGNORE INTO host_software_installed_paths (host_id, software_id, installed_path) VALUES (?, ?, ?)`, + concurrentTestHosts[i].ID, sw.ID, fmt.Sprintf("/Applications/%s.app", sw.Name)) + if err != nil { + return err + } + } + return nil + }) + } + + for i := range concurrentHosts { + go func(idx int) { + defer wg.Done() + h := concurrentTestHosts[idx] + newSwCopy := slices.Clone(newSoftware) + + startReplace := time.Now() + _, err := ds.UpdateHostSoftware(ctx, h.ID, newSwCopy) + elapsedReplace := time.Since(startReplace) + t.Logf(" Host %d replacement took: %s", idx, elapsedReplace) + if err != nil { + t.Errorf(" Host %d replacement error: %v", idx, err) + } + }(i) + } + wg.Wait() +} diff --git a/server/datastore/mysql/testing_utils_test.go b/server/datastore/mysql/testing_utils_test.go index 087592c810..0aa677a6d1 100644 --- a/server/datastore/mysql/testing_utils_test.go +++ b/server/datastore/mysql/testing_utils_test.go @@ -464,6 +464,9 @@ func TruncateTables(t testing.TB, ds *Datastore, tables ...string) { "DELETE FROM software_categories WHERE team_id != 0") require.NoError(t, err) testing_utils.TruncateTables(t, ds.writer(context.Background()), ds.logger, nonEmptyTables, tables...) + // Clear the in-process software title cache so it doesn't retain entries + // for titles that were just truncated from the database. + ds.clearKnownSoftwareTitleKeys() } // this is meant to be used for debugging/testing that statement uses an efficient