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,