perf(ingestor): group commit observation INSERTs by time window (M1, refs #1115) (#1117)

## 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 <bot@corescope.local>

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 {

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()
 }

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)
+	}
+}

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()
 

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,