port(upstream#1958): hash migration no longer reports success it never achieved

cmd/server/hash_migrate.go

@@ -10,9 +10,23 @@ import (
 // both the DB and the in-memory byHash index.  The migration is idempotent:
 // once all hashes match the current formula it completes instantly.
 func migrateContentHashesAsync(store *PacketStore, batchSize int, yieldDuration time.Duration) {
+	// #1856: every DB failure below continues to the next batch, so the loop
+	// always reaches the deferred completion. Setting the flag unconditionally
+	// therefore reported success after migrating nothing, which is exactly what
+	// happens on the read-only handle the server holds since #1283: begin,
+	// prepare and commit all fail, every batch is skipped, and /api/stats then
+	// answers hashMigrationComplete: true.
+	failedBatches := 0
 	defer func() {
 		if r := recover(); r != nil {
 			log.Printf("[hash-migrate] panic recovered: %v", r)
+			failedBatches++
+		}
+		if failedBatches > 0 {
+			log.Printf("[hash-migrate] INCOMPLETE: %d batch(es) could not be written; "+
+				"hashMigrationComplete stays false. On a read-only DB handle this is expected "+
+				"and the migration belongs in the ingestor (#1856).", failedBatches)
+			return
 		}
 		store.hashMigrationComplete.Store(true)
 	}()
@@ -57,12 +71,14 @@ func migrateContentHashesAsync(store *PacketStore, batchSize int, yieldDuration
 		dbTx, err := store.db.conn.Begin()
 		if err != nil {
 			log.Printf("[hash-migrate] begin tx: %v", err)
+			failedBatches++
 			continue
 		}
 		stmt, err := dbTx.Prepare("UPDATE transmissions SET hash = ? WHERE id = ?")
 		if err != nil {
 			log.Printf("[hash-migrate] prepare: %v", err)
 			dbTx.Rollback()
+			failedBatches++
 			continue
 		}
 
@@ -83,6 +99,7 @@ func migrateContentHashesAsync(store *PacketStore, batchSize int, yieldDuration
 
 		if err := dbTx.Commit(); err != nil {
 			log.Printf("[hash-migrate] commit: %v", err)
+			failedBatches++
 			continue
 		}
 

cmd/server/hash_migrate_test.go

@@ -76,3 +76,47 @@ func TestMigrateContentHashesAsync_NoOp(t *testing.T) {
 		t.Error("hash should remain in index")
 	}
 }
+
+// #1856: a migration that could not write anything must not report completion.
+//
+// In production the server holds a mode=ro handle (#1283), so Begin, Prepare and
+// Commit all fail, every batch takes a `continue`, and the loop reaches the
+// deferred completion having migrated nothing. Before this fix the flag was set
+// unconditionally there, so /api/stats answered hashMigrationComplete: true
+// after doing no work at all.
+//
+// Closing the handle stands in for the read-only one: it is deterministic and it
+// exercises the identical failure path (Begin returns an error, batch skipped).
+func TestMigrateContentHashesAsyncDoesNotClaimCompletionWhenWritesFail(t *testing.T) {
+	db := setupTestDBv2(t)
+	store := NewPacketStore(db, nil)
+
+	rawHex := "0A00D69FD7A5A7475DB07337749AE61FA53A4788E976"
+	wrongHash := "deadbeef12345678"
+	if _, err := db.conn.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type)
+		VALUES (?, ?, datetime('now'), 0, 2)`, rawHex, wrongHash); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.Load(); err != nil {
+		t.Fatal(err)
+	}
+	if store.byHash[wrongHash] == nil {
+		t.Fatal("expected packet under the wrong hash before migration")
+	}
+
+	// Make every write fail, the way a read-only handle does in production.
+	if err := db.conn.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	migrateContentHashesAsync(store, 100, time.Millisecond)
+
+	if store.hashMigrationComplete.Load() {
+		t.Error("hashMigrationComplete must stay false when no batch could be written; " +
+			"reporting true here is what #1856 called self-reported success")
+	}
+	if store.byHash[wrongHash] == nil {
+		t.Error("the in-memory index must be left alone when the DB write failed, " +
+			"otherwise memory and disk disagree")
+	}
+}