From 45f2607f75da3f99adba59a1152286424dfb5ec0 Mon Sep 17 00:00:00 2001 From: Kpa-clawbot Date: Tue, 5 May 2026 16:38:43 -0700 Subject: [PATCH] perf(ingestor): group commit observation INSERTs by time window (M1, refs #1115) (#1117) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Implements **M1 from #1115**: batches observation/transmission INSERTs into a single SQLite `BEGIN/COMMIT` window instead of fsyncing per packet. At ~250 obs/sec this drops WAL fsync rate from ~20/s to ~1/s and eliminates the `obs-persist skipped` / `SQLITE_BUSY` log spam that the issue documents. This is a **partial fix** — it ships the group-commit mechanism. Acceptance items 6–7 (measured fsync rate / measured `obs-persist skipped` rate at staging steady-state) require post-deploy observation, and M2 (per-`tx_hash` observation buffering) is intentionally deferred. The issue stays open for the user to verify on staging. > Partial fix for #1115 — does not auto-close. Refs #1115. ## Mechanism - `Store` gains an active `*sql.Tx`, `pendingRows` counter, `gcMu`, and the `groupCommitMs` / `groupCommitMaxRows` knobs. `SetGroupCommit(ms, maxRows)` enables the mode; `FlushGroupTx()` commits the in-flight tx. - `InsertTransmission` lazily opens a tx on the first call after each flush, then issues all writes through `tx.Stmt()` bindings of the existing prepared statements. With `MaxOpenConns(1)` the connection is already serialized; `gcMu` serializes group-commit state without contention. - A goroutine in `cmd/ingestor/main.go` calls `FlushGroupTx()` every `groupCommitMs` ms. `pendingRows >= groupCommitMaxRows` triggers an eager flush. `Close()` flushes before the WAL checkpoint so no rows are lost on graceful shutdown. - `groupCommitMs == 0` short-circuits to the legacy per-call auto-commit path (statements bound to `s.db`, no tx) — current behavior preserved byte-for-byte for operators who opt out. ## Config Two new optional fields (ingestor-only), both documented in `config.example.json`: | Field | Default | Effect | |---|---|---| | `groupCommitMs` | `1000` | Flush window in ms. `0` disables batching (legacy per-packet auto-commit). | | `groupCommitMaxRows` | `1000` | Safety cap; when exceeded the queue flushes immediately to bound memory and the crash-loss window. | No DB schema change. No required config change on upgrade. ## Tests (TDD red → green visible in commits) `cmd/ingestor/group_commit_test.go` — three assertions, written first as the red commit: - `TestGroupCommit_BatchesInsertsIntoOneTx` — 50 `InsertTransmission` calls inside a wide window produce **0** commits until `FlushGroupTx`, then exactly **1**; all 50 rows visible after flush. (This is the spec's "50 observations → 1 SQLite write transaction" assertion.) - `TestGroupCommit_Disabled` — `groupCommitMs=0` keeps every insert immediately visible and `GroupCommitFlushes` never advances. (Spec's "groupCommitMs=0 reverts to per-packet behavior" assertion.) - `TestGroupCommit_MaxRowsForcesEarlyFlush` — cap=3, 7 inserts → 2 auto-flushes from the cap + 1 final manual flush = 3 total. Red commit: `e2b0370` (stubs `SetGroupCommit` / `FlushGroupTx` so the tests compile and fail on **assertions**, not import errors). Green commit: `73f3559`. Full ingestor suite (`go test ./...` in `cmd/ingestor`) stays green, ~49 s. ## Performance This PR is the perf change itself. Local micro-test (the new `TestGroupCommit_BatchesInsertsIntoOneTx`) shows the structural property: 50 inserts → 1 commit. The fsync-rate measurement called out in the M1 acceptance criteria (`~20/s → ~1/s` at 250 obs/sec) requires staging deployment to confirm — that's the remaining open item that keeps #1115 open after this merges. No hot-path regressions: when `groupCommitMs > 0` we acquire one mutex per insert (uncontended in the steady state — the connection was already single-threaded via `MaxOpenConns(1)`). When `groupCommitMs == 0` the code path is identical to before plus one nil-tx check. ## What this PR does NOT do (per spec) - Does not collapse "30 observations of one packet" into 1 row write — that's M2. - Does not eliminate dual-writer contention with `cmd/server`'s `resolved_path` writes. - Does not change observation ordering or live broadcast latency. --------- Co-authored-by: corescope-bot --- cmd/ingestor/config.go | 28 ++++++ cmd/ingestor/db.go | 140 +++++++++++++++++++++++++-- cmd/ingestor/group_commit_test.go | 155 ++++++++++++++++++++++++++++++ cmd/ingestor/main.go | 20 ++++ config.example.json | 4 + 5 files changed, 341 insertions(+), 6 deletions(-) create mode 100644 cmd/ingestor/group_commit_test.go diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index 0429b23a..8510d8bd 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -72,6 +72,17 @@ type Config struct { // no UpsertObserver, no observations, no metrics. ObserverBlacklist []string `json:"observerBlacklist,omitempty"` + // GroupCommitMs controls observation INSERT batching (#1115 M1). When > 0, + // the ingestor wraps pending INSERTs into a single BEGIN/COMMIT and flushes + // every GroupCommitMs milliseconds. When 0, every InsertTransmission commits + // individually (legacy per-packet behavior). Default applied at runtime: 1000. + GroupCommitMs *int `json:"groupCommitMs,omitempty"` + + // GroupCommitMaxRows is a safety cap on pending rows in the group-commit + // queue. When exceeded, the queue flushes immediately to bound memory and + // the crash window. Default applied at runtime: 1000. + GroupCommitMaxRows *int `json:"groupCommitMaxRows,omitempty"` + // obsBlacklistSetCached is the lazily-built lowercase set for O(1) lookups. obsBlacklistSetCached map[string]bool obsBlacklistOnce sync.Once @@ -136,6 +147,23 @@ func (c *Config) MetricsSampleInterval() int { return 300 } +// GroupCommitMsOrDefault returns the configured groupCommitMs or 1000 if unset. +// A value of 0 explicitly disables group commit (per-packet auto-commit). +func (c *Config) GroupCommitMsOrDefault() int { + if c == nil || c.GroupCommitMs == nil { + return 1000 + } + return *c.GroupCommitMs +} + +// GroupCommitMaxRowsOrDefault returns the configured cap or 1000 if unset. +func (c *Config) GroupCommitMaxRowsOrDefault() int { + if c == nil || c.GroupCommitMaxRows == nil || *c.GroupCommitMaxRows <= 0 { + return 1000 + } + return *c.GroupCommitMaxRows +} + // MetricsRetentionDays returns configured metrics retention or 30 days default. func (c *Config) MetricsRetentionDays() int { if c.Retention != nil && c.Retention.MetricsDays > 0 { diff --git a/cmd/ingestor/db.go b/cmd/ingestor/db.go index a71fb61b..edeee1fc 100644 --- a/cmd/ingestor/db.go +++ b/cmd/ingestor/db.go @@ -25,6 +25,64 @@ type DBStats struct { ObserverUpserts atomic.Int64 WriteErrors atomic.Int64 SignatureDrops atomic.Int64 + GroupCommitFlushes atomic.Int64 +} + +// SetGroupCommit configures group-commit batching for InsertTransmission. +// When ms > 0, observation/transmission INSERTs are queued inside a single +// BEGIN/COMMIT and flushed every ms milliseconds (or earlier if the pending +// row count exceeds maxRows). When ms == 0 every InsertTransmission commits +// individually (legacy behavior). +// +// Safe to call at any time; an in-flight transaction is committed before +// the new mode takes effect. +func (s *Store) SetGroupCommit(ms int, maxRows int) { + if maxRows <= 0 { + maxRows = 1000 + } + s.gcMu.Lock() + // Flush any open tx under the old config before swapping. + if s.activeTx != nil { + _ = s.activeTx.Commit() + s.activeTx = nil + s.pendingRows = 0 + s.Stats.GroupCommitFlushes.Add(1) + } + s.groupCommitMs = ms + s.groupCommitMaxRows = maxRows + s.gcMu.Unlock() +} + +// FlushGroupTx commits any pending grouped INSERTs immediately. Safe to call +// when group commit is disabled (no-op). Safe to call from a separate +// goroutine — serialized via gcMu. +func (s *Store) FlushGroupTx() error { + s.gcMu.Lock() + defer s.gcMu.Unlock() + return s.flushLocked() +} + +// flushLocked commits the active tx if any. Caller must hold gcMu. +func (s *Store) flushLocked() error { + if s.activeTx == nil { + return nil + } + err := s.activeTx.Commit() + s.activeTx = nil + s.pendingRows = 0 + s.Stats.GroupCommitFlushes.Add(1) + if err != nil { + s.Stats.WriteErrors.Add(1) + return fmt.Errorf("group commit: %w", err) + } + return nil +} + +// GroupCommitMs returns the configured flush window in ms (0 = disabled). +func (s *Store) GroupCommitMs() int { + s.gcMu.Lock() + defer s.gcMu.Unlock() + return s.groupCommitMs } // Store wraps the SQLite database for packet ingestion. @@ -46,6 +104,17 @@ type Store struct { sampleIntervalSec int backfillWg sync.WaitGroup + + // Group-commit state (#1115 M1). When groupCommitMs > 0, observation + // INSERTs are queued inside a single BEGIN/COMMIT and flushed every + // groupCommitMs ms (via the ingestor's flusher goroutine) or earlier + // when groupCommitMaxRows is exceeded. When groupCommitMs == 0 every + // InsertTransmission commits individually (legacy behavior). + gcMu sync.Mutex + groupCommitMs int + groupCommitMaxRows int + activeTx *sql.Tx + pendingRows int } // OpenStore opens or creates a SQLite DB at the given path, applying the @@ -595,6 +664,12 @@ func (s *Store) prepareStatements() error { // InsertTransmission inserts a decoded packet into transmissions + observations. // Returns true if a new transmission was created (not a duplicate hash). +// +// When group-commit is enabled (via SetGroupCommit with ms > 0), all writes +// are issued against an in-flight *sql.Tx that is committed by the ingestor's +// flusher goroutine (or eagerly when pendingRows exceeds groupCommitMaxRows). +// When disabled (ms == 0) every call commits immediately via the prepared +// statements bound to s.db (legacy behavior). func (s *Store) InsertTransmission(data *PacketData) (bool, error) { hash := data.Hash if hash == "" { @@ -606,29 +681,67 @@ func (s *Store) InsertTransmission(data *PacketData) (bool, error) { now = time.Now().UTC().Format(time.RFC3339) } + s.gcMu.Lock() + defer s.gcMu.Unlock() + + // Stmt resolvers — either bare prepared stmt (auto-commit) or tx-bound. + var ( + stmtGetTx, stmtUpdFS, stmtInsTx, stmtGetObs, stmtUpdObs, stmtInsObs *sql.Stmt + ) + if s.groupCommitMs > 0 { + if s.activeTx == nil { + tx, err := s.db.Begin() + if err != nil { + s.Stats.WriteErrors.Add(1) + return false, fmt.Errorf("group commit begin: %w", err) + } + s.activeTx = tx + s.pendingRows = 0 + } + stmtGetTx = s.activeTx.Stmt(s.stmtGetTxByHash) + stmtUpdFS = s.activeTx.Stmt(s.stmtUpdateTxFirstSeen) + stmtInsTx = s.activeTx.Stmt(s.stmtInsertTransmission) + stmtGetObs = s.activeTx.Stmt(s.stmtGetObserverRowid) + stmtUpdObs = s.activeTx.Stmt(s.stmtUpdateObserverLastSeen) + stmtInsObs = s.activeTx.Stmt(s.stmtInsertObservation) + } else { + stmtGetTx = s.stmtGetTxByHash + stmtUpdFS = s.stmtUpdateTxFirstSeen + stmtInsTx = s.stmtInsertTransmission + stmtGetObs = s.stmtGetObserverRowid + stmtUpdObs = s.stmtUpdateObserverLastSeen + stmtInsObs = s.stmtInsertObservation + } + var txID int64 isNew := false // Check for existing transmission var existingID int64 var existingFirstSeen string - err := s.stmtGetTxByHash.QueryRow(hash).Scan(&existingID, &existingFirstSeen) + err := stmtGetTx.QueryRow(hash).Scan(&existingID, &existingFirstSeen) if err == nil { // Existing transmission txID = existingID if now < existingFirstSeen { - _, _ = s.stmtUpdateTxFirstSeen.Exec(now, txID) + _, _ = stmtUpdFS.Exec(now, txID) } } else { // New transmission isNew = true - result, err := s.stmtInsertTransmission.Exec( + result, err := stmtInsTx.Exec( data.RawHex, hash, now, data.RouteType, data.PayloadType, data.PayloadVersion, data.DecodedJSON, nilIfEmpty(data.ChannelHash), ) if err != nil { s.Stats.WriteErrors.Add(1) + // Rollback in-flight tx so we don't leave it dangling. + if s.activeTx != nil { + _ = s.activeTx.Rollback() + s.activeTx = nil + s.pendingRows = 0 + } return false, fmt.Errorf("insert transmission: %w", err) } txID, _ = result.LastInsertId() @@ -643,12 +756,12 @@ func (s *Store) InsertTransmission(data *PacketData) (bool, error) { var observerIdx *int64 if data.ObserverID != "" { var rowid int64 - err := s.stmtGetObserverRowid.QueryRow(data.ObserverID).Scan(&rowid) + err := stmtGetObs.QueryRow(data.ObserverID).Scan(&rowid) if err == nil { observerIdx = &rowid // Update observer last_seen and last_packet_at on every packet to prevent // low-traffic observers from appearing offline (#463) - _, _ = s.stmtUpdateObserverLastSeen.Exec(now, now, rowid) + _, _ = stmtUpdObs.Exec(now, now, rowid) } } @@ -658,7 +771,7 @@ func (s *Store) InsertTransmission(data *PacketData) (bool, error) { epochTs = t.Unix() } - _, err = s.stmtInsertObservation.Exec( + _, err = stmtInsObs.Exec( txID, observerIdx, data.Direction, data.SNR, data.RSSI, data.Score, data.PathJSON, epochTs, nilIfEmpty(data.RawHex), @@ -670,6 +783,17 @@ func (s *Store) InsertTransmission(data *PacketData) (bool, error) { s.Stats.ObservationsInserted.Add(1) } + // Group-commit accounting: count this insert and flush early if we hit + // the row cap. Counts pending rows (transmission + observation pairs). + if s.activeTx != nil { + s.pendingRows++ + if s.pendingRows >= s.groupCommitMaxRows { + if err := s.flushLocked(); err != nil { + log.Printf("[db] group commit (max-rows) flush failed: %v", err) + } + } + } + return isNew, nil } @@ -793,6 +917,10 @@ func (s *Store) UpsertObserver(id, name, iata string, meta *ObserverMeta) error // Close checkpoints the WAL and closes the database. func (s *Store) Close() error { s.backfillWg.Wait() + // Flush any pending grouped INSERTs before checkpoint/close (#1115). + if err := s.FlushGroupTx(); err != nil { + log.Printf("[db] close: group commit flush: %v", err) + } s.Checkpoint() return s.db.Close() } diff --git a/cmd/ingestor/group_commit_test.go b/cmd/ingestor/group_commit_test.go new file mode 100644 index 00000000..67edc615 --- /dev/null +++ b/cmd/ingestor/group_commit_test.go @@ -0,0 +1,155 @@ +package main + +import ( + "fmt" + "testing" +) + +// makePacket returns a minimal valid PacketData with a unique hash so +// each call is treated as a distinct transmission by InsertTransmission. +func makePacket(i int) *PacketData { + snr := 1.0 + rssi := -90.0 + return &PacketData{ + RawHex: fmt.Sprintf("AABB%04X", i), + Timestamp: "2026-05-01T00:00:00Z", + ObserverID: "obsGC", + Hash: fmt.Sprintf("gchash%010d", i), + RouteType: 2, + PayloadType: 2, + PayloadVersion: 0, + PathJSON: "[]", + DecodedJSON: `{"type":"TXT_MSG"}`, + SNR: &snr, + RSSI: &rssi, + } +} + +// TestGroupCommit_BatchesInsertsIntoOneTx verifies M1 behavior: with +// groupCommitMs > 0, 50 InsertTransmission calls should produce ZERO +// commits until FlushGroupTx is called, then exactly 1 commit. +func TestGroupCommit_BatchesInsertsIntoOneTx(t *testing.T) { + s, err := OpenStore(tempDBPath(t)) + if err != nil { + t.Fatal(err) + } + defer s.Close() + + if err := s.UpsertObserver("obsGC", "GC Observer", "SJC", nil); err != nil { + t.Fatal(err) + } + + // Enable group commit with a wide window so the test ticker doesn't fire. + s.SetGroupCommit(60_000, 1000) + + startFlushes := s.Stats.GroupCommitFlushes.Load() + + for i := 0; i < 50; i++ { + if _, err := s.InsertTransmission(makePacket(i)); err != nil { + t.Fatalf("insert %d: %v", i, err) + } + } + + // Before flush, no commits should have occurred. (max=1000, count=50.) + if got := s.Stats.GroupCommitFlushes.Load() - startFlushes; got != 0 { + t.Fatalf("flushes before manual flush: got %d, want 0", got) + } + + // Manual flush — exactly one commit for all 50 inserts. + if err := s.FlushGroupTx(); err != nil { + t.Fatalf("FlushGroupTx: %v", err) + } + if got := s.Stats.GroupCommitFlushes.Load() - startFlushes; got != 1 { + t.Fatalf("flushes after manual flush: got %d, want 1", got) + } + + // All 50 rows must be visible after commit. + var n int + if err := s.db.QueryRow("SELECT COUNT(*) FROM transmissions WHERE hash LIKE 'gchash%'").Scan(&n); err != nil { + t.Fatal(err) + } + if n != 50 { + t.Fatalf("transmissions after flush: got %d, want 50", n) + } + if err := s.db.QueryRow("SELECT COUNT(*) FROM observations").Scan(&n); err != nil { + t.Fatal(err) + } + if n != 50 { + t.Fatalf("observations after flush: got %d, want 50", n) + } +} + +// TestGroupCommit_Disabled verifies that with groupCommitMs == 0, every +// InsertTransmission commits immediately (current behavior preserved) and +// the GroupCommitFlushes counter never advances. +func TestGroupCommit_Disabled(t *testing.T) { + s, err := OpenStore(tempDBPath(t)) + if err != nil { + t.Fatal(err) + } + defer s.Close() + + if err := s.UpsertObserver("obsGC", "GC Observer", "SJC", nil); err != nil { + t.Fatal(err) + } + + // Explicitly disable. + s.SetGroupCommit(0, 1000) + + startFlushes := s.Stats.GroupCommitFlushes.Load() + + for i := 0; i < 5; i++ { + if _, err := s.InsertTransmission(makePacket(i)); err != nil { + t.Fatalf("insert %d: %v", i, err) + } + // Each insert is immediately visible — no flush required. + var n int + if err := s.db.QueryRow("SELECT COUNT(*) FROM transmissions WHERE hash LIKE 'gchash%'").Scan(&n); err != nil { + t.Fatal(err) + } + if n != i+1 { + t.Fatalf("after insert %d: got %d transmissions, want %d", i, n, i+1) + } + } + + if got := s.Stats.GroupCommitFlushes.Load() - startFlushes; got != 0 { + t.Fatalf("flushes with group commit disabled: got %d, want 0", got) + } +} + +// TestGroupCommit_MaxRowsForcesEarlyFlush verifies that exceeding the +// row cap triggers an immediate flush even before the ticker fires. +func TestGroupCommit_MaxRowsForcesEarlyFlush(t *testing.T) { + s, err := OpenStore(tempDBPath(t)) + if err != nil { + t.Fatal(err) + } + defer s.Close() + + if err := s.UpsertObserver("obsGC", "GC Observer", "SJC", nil); err != nil { + t.Fatal(err) + } + + // Window large; cap small (3) so the 4th insert should flush. + s.SetGroupCommit(60_000, 3) + + startFlushes := s.Stats.GroupCommitFlushes.Load() + + for i := 0; i < 7; i++ { + if _, err := s.InsertTransmission(makePacket(i)); err != nil { + t.Fatalf("insert %d: %v", i, err) + } + } + + // 7 inserts with cap 3 → 2 auto-flushes (after 3 and after 6); 1 still pending. + if got := s.Stats.GroupCommitFlushes.Load() - startFlushes; got != 2 { + t.Fatalf("auto-flushes: got %d, want 2", got) + } + + if err := s.FlushGroupTx(); err != nil { + t.Fatal(err) + } + if got := s.Stats.GroupCommitFlushes.Load() - startFlushes; got != 3 { + t.Fatalf("flushes after final manual flush: got %d, want 3", got) + } +} diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index 3ad34424..24548220 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -57,6 +57,26 @@ func main() { defer store.Close() log.Printf("SQLite opened: %s", cfg.DBPath) + // #1115 M1: enable group commit and start a flusher goroutine. When + // groupCommitMs == 0 the store falls back to per-call auto-commit and + // the ticker is a cheap no-op (FlushGroupTx is a noop with no active tx). + gcMs := cfg.GroupCommitMsOrDefault() + gcMax := cfg.GroupCommitMaxRowsOrDefault() + store.SetGroupCommit(gcMs, gcMax) + if gcMs > 0 { + log.Printf("group commit: window=%dms maxRows=%d", gcMs, gcMax) + gcTicker := time.NewTicker(time.Duration(gcMs) * time.Millisecond) + go func() { + for range gcTicker.C { + if err := store.FlushGroupTx(); err != nil { + log.Printf("[db] group commit flush: %v", err) + } + } + }() + } else { + log.Printf("group commit: disabled (per-packet auto-commit)") + } + // Async backfill: path_json from raw_hex (#888) — must not block MQTT startup store.BackfillPathJSONAsync() diff --git a/config.example.json b/config.example.json index 80ac0a55..92358acc 100644 --- a/config.example.json +++ b/config.example.json @@ -5,6 +5,10 @@ "_comment_nodeBlacklist": "Public keys of nodes to hide from all API responses. Use for trolls, offensive names, or nodes reporting false data that operators refuse to fix.", "observerIATAWhitelist": [], "_comment_observerIATAWhitelist": "Global IATA region whitelist. When non-empty, only observers whose IATA code (from MQTT topic) matches are processed. Case-insensitive. Empty = allow all. Unlike per-source iataFilter, this applies across all MQTT sources.", + "groupCommitMs": 1000, + "_comment_groupCommitMs": "Ingestor only (#1115 M1). Window in milliseconds for batching observation INSERTs into a single SQLite transaction. Default 1000 (1s). Set to 0 to disable batching and revert to per-packet auto-commit (legacy behavior). Trade-off: up to this many ms of delay before observations are queryable via SQL (live WebSocket broadcast is unaffected). Reduces WAL fsync rate from ~per-packet to ~1/window, eliminating SQLITE_BUSY/obs-persist-skipped log spam at high ingest rates.", + "groupCommitMaxRows": 1000, + "_comment_groupCommitMaxRows": "Ingestor only (#1115 M1). Safety cap on pending rows in the group-commit queue. When exceeded, the queue flushes immediately even if the time window has not elapsed. Bounds memory use and the crash-loss window. Default 1000.", "retention": { "nodeDays": 7, "observerDays": 14,