fix(state): three history writers hold their mutex until rollback ends (#1845)

* fix(state): three history writers hold their mutex until rollback ends

The control loop's history write, the SQLite history import and the
plain archive import held historyWriteMu around a transaction begun on
the *sql.DB. When that context ended, database/sql could report the Tx
done before SQLite had rolled it back, and the mutex went to the next
writer early. They now begin the transaction on a connection of their
own and close it before unlocking, as the batch and maintenance paths
already do.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* chore(changeset): the control loop already wrote on its own connection

Review note: the control loop writes through recordHistoryBatches,
which used a Conn already. recordHistoryBatch serves RecordTick*, which
only tests call, so the changeset no longer names the control loop.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

Author: frahlg

.changeset/history-writer-own-connection.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+When the import of older history or of plain archive files is stopped or runs out of time, other history writes now wait until SQLite has rolled it back. Before, the next write could start too early and then wait or fail as busy. No data was lost either way.

go/internal/state/history_migration.go

@@ -299,11 +299,11 @@ func (s *Store) mergeHistoricalSamples(ctx context.Context, samples []Sample, re
 	}
 	s.historyWriteMu.Lock()
 	defer s.historyWriteMu.Unlock()
-	tx, err := s.history.BeginTx(ctx, nil)
+	tx, end, err := s.beginHistoryWrite(ctx)
 	if err != nil {
 		return err
 	}
-	defer tx.Rollback()
+	defer end()
 	if err := s.insertSamplesAndHours(ctx, tx, rs); err != nil {
 		return err
 	}

go/internal/state/history_plain.go

@@ -334,11 +334,11 @@ func (s *Store) writeHourBuckets(ctx context.Context, hours map[string]*namedHou
 		return err
 	}
 	defer s.historyWriteMu.Unlock()
-	tx, err := s.history.BeginTx(ctx, nil)
+	tx, end, err := s.beginHistoryWrite(ctx)
 	if err != nil {
 		return err
 	}
-	defer tx.Rollback()
+	defer end()
 	var imported bool
 	if err := tx.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM history_migrations WHERE name=?)`, receipt).Scan(&imported); err != nil {
 		return err

go/internal/state/history_rollback_test.go

@@ -182,8 +182,10 @@ func isHistoryTestWrite(query string) bool {
 	return strings.HasPrefix(q, "INSERT ") || strings.HasPrefix(q, "UPDATE ") || strings.HasPrefix(q, "DELETE ")
 }
 
-func TestHistoryMaintenanceWaitsForCancelledRollback(t *testing.T) {
-	for _, kind := range []string{"archive", "backfill", "completion"} {
+// Each writer that holds historyWriteMu around a transaction keeps it until
+// SQLite has rolled a cancelled transaction back.
+func TestHistoryWriterWaitsForCancelledRollback(t *testing.T) {
+	for _, kind := range []string{"archive", "backfill", "completion", "tick", "cold-hours", "import"} {
 		t.Run(kind, func(t *testing.T) {
 			s := freshStore(t)
 			if kind == "backfill" {
@@ -219,6 +221,14 @@ func TestHistoryMaintenanceWaitsForCancelledRollback(t *testing.T) {
 				case "backfill":
 					_, err := s.backfillSeriesHoursSnapshot(ctx, 64)
 					return err
+				case "tick":
+					_, err := s.recordHistoryBatch(ctx, "", "", &HistoryPoint{TsMs: seriesHourMs}, nil, nil, 0)
+					return err
+				case "cold-hours":
+					minute := BucketSummary{StartMS: seriesHourMs, ResolutionMS: ArchiveResolutionMS, FirstMS: seriesHourMs, LastMS: seriesHourMs, N: 1, Sum: 1200, Min: 1200, Max: 1200, Last: 1200}
+					return s.writeHourBuckets(ctx, map[string]*namedHour{"meter": {driver: "meter", metric: "pv_w", hour: seriesHourMs, sources: []BucketSummary{minute}}}, "plain-archive:rollback-test")
+				case "import":
+					return s.mergeHistoricalSamples(ctx, []Sample{{Driver: "meter", Metric: "pv_w", TsMs: seriesHourMs + 1000, Value: 1300}}, nil)
 				default:
 					return s.ensureSeriesHours(ctx)
 				}
@@ -228,23 +238,23 @@ func TestHistoryMaintenanceWaitsForCancelledRollback(t *testing.T) {
 			d.armed.Store(true)
 			done := make(chan error, 1)
 			go func() { done <- run(ctx) }()
-			// Cancel once the maintenance write is under way, so the rollback
-			// held below is that write's. A deadline could land earlier, in a
+			// Cancel once the write is under way, so the rollback held below
+			// is that write's. A deadline could land earlier, in a
 			// read that holds no writer mutex yet.
 			select {
 			case <-d.blocked:
 			case err := <-done:
-				t.Fatalf("maintenance ended before its write: %v", err)
+				t.Fatalf("the write ended before it wrote: %v", err)
 			case <-time.After(5 * time.Second):
-				t.Fatal("maintenance did not reach its write")
+				t.Fatal("the write did not start")
 			}
 			cancel()
 			select {
 			case <-d.started:
 			case err := <-done:
-				t.Fatalf("maintenance ended before rollback: %v", err)
+				t.Fatalf("the write ended before rollback: %v", err)
 			case <-time.After(5 * time.Second):
-				t.Fatal("maintenance did not start rollback")
+				t.Fatal("the write did not start rollback")
 			}
 			waitCtx, stop := context.WithTimeout(context.Background(), 50*time.Millisecond)
 			lockErr := lockContext(waitCtx, s.historyWriteMu.TryLock)
@@ -256,15 +266,15 @@ func TestHistoryMaintenanceWaitsForCancelledRollback(t *testing.T) {
 			select {
 			case err = <-done:
 			case <-time.After(5 * time.Second):
-				t.Fatal("maintenance did not finish after rollback")
+				t.Fatal("the write did not finish after rollback")
 			}
 			if lockErr == nil {
-				t.Fatal("maintenance released the writer mutex before SQLite rollback finished")
+				t.Fatal("the write released the writer mutex before SQLite rollback finished")
 			}
 			// The archive write also has its own 1 s limit, which a stalled
 			// test goroutine can let expire before it cancels.
 			if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
-				t.Fatalf("cancelled maintenance: %v", err)
+				t.Fatalf("cancelled write: %v", err)
 			}
 			if err := run(context.Background()); err != nil {
 				t.Fatalf("retry after rollback: %v", err)

go/internal/state/store_ts.go

@@ -393,11 +393,11 @@ func (s *Store) recordHistoryBatch(ctx context.Context, batchID, payloadHash str
 
 	s.historyWriteMu.Lock()
 	defer s.historyWriteMu.Unlock()
-	tx, err := s.history.BeginTx(ctx, nil)
+	tx, end, err := s.beginHistoryWrite(ctx)
 	if err != nil {
 		return 0, err
 	}
-	defer tx.Rollback()
+	defer end()
 	seq, skipped, err := s.writeHistoryBatchTx(ctx, tx, batchID, payloadHash, p, rs, observations)
 	if err != nil {
 		return 0, err
@@ -413,6 +413,27 @@ func (s *Store) recordHistoryBatch(ctx context.Context, batchID, payloadHash str
 	return seq, nil
 }
 
+// beginHistoryWrite begins a transaction on a connection of its own, for a
+// caller that holds historyWriteMu and calls end before it unlocks. A
+// cancelled Tx may report ErrTxDone before its automatic rollback releases
+// SQLite's writer. Closing the connection waits for that rollback, so the
+// next writer cannot read a snapshot and then fail its upgrade with BUSY.
+func (s *Store) beginHistoryWrite(ctx context.Context) (tx *sql.Tx, end func(), err error) {
+	conn, err := s.history.Conn(ctx)
+	if err != nil {
+		return nil, nil, err
+	}
+	tx, err = conn.BeginTx(ctx, nil)
+	if err != nil {
+		_ = conn.Close()
+		return nil, nil, err
+	}
+	return tx, func() {
+		_ = tx.Rollback()
+		_ = conn.Close()
+	}, nil
+}
+
 func ackHistoryReceipts(ctx context.Context, tx *sql.Tx, seq, acknowledgedSequence int64) error {
 	if acknowledgedSequence <= 0 || seq <= 0 {
 		return nil