Schema & robustness hardening: contents immutability triggers (v21) + #114 cluster

cmd/squirrel/sync.go

@@ -41,7 +41,7 @@ func newSyncCmd() *cobra.Command {
 	cmd.Flags().StringVar(&to, "to", "", "limit to this destination name (default: every destination declared on the volume)")
 	cmd.Flags().BoolVar(&shallow, "shallow", false, "skip BLAKE3 verification; trust rclone's default size+mtime comparison")
 	cmd.Flags().BoolVar(&dryRun, "dry-run", false, "preview rclone actions without transferring; no runs row is written")
-	cmd.Flags().BoolVar(&initDst, "init", false, "bootstrap a .squirrel-volume marker at the destination on first sync (refused subsequently if the marker mismatches)")
+	cmd.Flags().BoolVar(&initDst, "init", false, "authorise first-use destination bootstrap: write a .squirrel-volume marker, or create a kopia repository when connect finds none (refused without --init so a typo or outage can't mint a fresh empty target)")
 	return cmd
 }
 

store/destination_run_ids_test.go

@@ -22,6 +22,15 @@ func TestMigrateV18ToV19AddsVerifyMethod(t *testing.T) {
 		`CREATE TABLE schema_version (version INTEGER NOT NULL PRIMARY KEY)`,
 		`CREATE TABLE volumes (id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE, path TEXT NOT NULL)`,
 		`CREATE TABLE nodes (id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE, endpoint TEXT, public_key_fingerprint TEXT)`,
+		// contents exists from v14 on; a real v18 DB carries it, and the
+		// v20→v21 triggers attach to it, so the minimal fixture must too.
+		`CREATE TABLE contents (
+			id             INTEGER PRIMARY KEY,
+			blake3         BLOB NOT NULL UNIQUE CHECK (length(blake3) = 32),
+			size_bytes     INTEGER NOT NULL,
+			origin_node_id INTEGER REFERENCES nodes(id),
+			origin_run_id  INTEGER
+		)`,
 		`CREATE TABLE destination_run_ids (
 			volume_id      INTEGER NOT NULL REFERENCES volumes(id),
 			destination    TEXT NOT NULL,

store/hookruns.go

@@ -3,6 +3,7 @@ package store
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"fmt"
 )
 
@@ -112,32 +113,58 @@ func (s *Store) BeginHookRun(ctx context.Context, spec HookRunSpec) (int64, erro
 	return id, nil
 }
 
+// isTerminalHookStatus reports whether status is one of the two terminal
+// hook states. A row in either must not be re-finalised by FinishHookRun.
+func isTerminalHookStatus(status string) bool {
+	return status == HookStatusSuccess || status == HookStatusFailed
+}
+
 // FinishHookRun records the terminal state of a hook run. exitCode is
 // stored as-is (pass an invalid sql.NullInt64 when the process produced
 // no code, e.g. spawn failure or timeout); errMsg is stored as NULL when
 // empty. Returns an error if id matches no row so a hook is never left
 // stuck in 'running'.
+//
+// Like FinishRun, the transition is guarded: a hook run already in a
+// terminal status is never re-finalised — the first terminal write wins
+// and FinishHookRun returns ErrAlreadyFinished (matchable via errors.Is)
+// without touching the row, so a double-finish bug or a buggy retry can't
+// silently rewrite the recorded status, exit code, and end timestamp. The
+// read and the update share one transaction so the check and the write
+// can't race.
 func (s *Store) FinishHookRun(ctx context.Context, id int64, status string, exitCode sql.NullInt64, errMsg string) error {
 	if status != HookStatusSuccess && status != HookStatusFailed {
 		return fmt.Errorf("FinishHookRun: status must be %q or %q, got %q", HookStatusSuccess, HookStatusFailed, status)
 	}
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("begin finish hook run %d: %w", id, err)
+	}
+	defer func() { _ = tx.Rollback() }()
+
+	var current string
+	switch err := tx.QueryRowContext(ctx, `SELECT status FROM hook_runs WHERE id = ?`, id).Scan(&current); {
+	case errors.Is(err, sql.ErrNoRows):
+		return fmt.Errorf("finish hook run %d: no such hook run", id)
+	case err != nil:
+		return fmt.Errorf("finish hook run %d read status: %w", id, err)
+	}
+	if isTerminalHookStatus(current) {
+		return fmt.Errorf("finish hook run %d (status %s): %w", id, current, ErrAlreadyFinished)
+	}
+
 	var errVal sql.NullString
 	if errMsg != "" {
 		errVal = sql.NullString{String: errMsg, Valid: true}
 	}
-	res, err := s.db.ExecContext(ctx, `
+	if _, err := tx.ExecContext(ctx, `
 		UPDATE hook_runs SET ended_at_ns = ?, status = ?, exit_code = ?, error = ?
 		WHERE id = ?
-	`, NowNs(), status, exitCode, errVal, id)
-	if err != nil {
+	`, NowNs(), status, exitCode, errVal, id); err != nil {
 		return fmt.Errorf("finish hook run %d: %w", id, err)
 	}
-	n, err := res.RowsAffected()
-	if err != nil {
-		return fmt.Errorf("finish hook run %d rows affected: %w", id, err)
-	}
-	if n == 0 {
-		return fmt.Errorf("finish hook run %d: no such hook run", id)
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("commit finish hook run %d: %w", id, err)
 	}
 	return nil
 }

store/hookruns_test.go

@@ -131,6 +131,45 @@ func TestFinishHookRunUnknownID(t *testing.T) {
 	}
 }
 
+// TestFinishHookRunRefusesTerminalRow: the first terminal write wins. A
+// second finish is refused with ErrAlreadyFinished and leaves the
+// recorded status, exit code, and end timestamp untouched (#114) — the
+// same first-write-wins guard FinishRun has.
+func TestFinishHookRunRefusesTerminalRow(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	vol, _ := s.CreateVolume(ctx, "v", "/tmp/v")
+	id, err := s.BeginHookRun(ctx, HookRunSpec{VolumeID: vol.ID, Trigger: HookTriggerInterval})
+	if err != nil {
+		t.Fatalf("BeginHookRun: %v", err)
+	}
+
+	firstExit := sql.NullInt64{Int64: 0, Valid: true}
+	if err := s.FinishHookRun(ctx, id, HookStatusSuccess, firstExit, ""); err != nil {
+		t.Fatalf("first FinishHookRun: %v", err)
+	}
+	before, err := s.hookRunByID(ctx, id)
+	if err != nil {
+		t.Fatalf("read back after first finish: %v", err)
+	}
+
+	err = s.FinishHookRun(ctx, id, HookStatusFailed, sql.NullInt64{Int64: 7, Valid: true}, "second finish")
+	if !errors.Is(err, ErrAlreadyFinished) {
+		t.Fatalf("second FinishHookRun err = %v, want ErrAlreadyFinished", err)
+	}
+
+	after, err := s.hookRunByID(ctx, id)
+	if err != nil {
+		t.Fatalf("read back after refused finish: %v", err)
+	}
+	if after.Status != HookStatusSuccess {
+		t.Fatalf("status = %q after refused second finish, want success", after.Status)
+	}
+	if after.ExitCode != before.ExitCode || after.EndedAtNs != before.EndedAtNs || after.Error != before.Error {
+		t.Fatalf("terminal row mutated by refused finish: before=%+v after=%+v", before, after)
+	}
+}
+
 func TestListHookRuns(t *testing.T) {
 	s := openTestStore(t)
 	ctx := context.Background()

store/migrations.go

@@ -10,7 +10,7 @@ import (
 )
 
 // SchemaVersion is the schema version this binary writes and reads.
-const SchemaVersion = 20
+const SchemaVersion = 21
 
 // freshSchemaBaseline is the version applied to a brand-new database. The
 // chain in `migrations` continues from here. v1 is no longer reachable from
@@ -56,6 +56,7 @@ func buildMigrations(mctx migrationCtx) []migration {
 		{version: 18, up: migrateV17ToV18},
 		{version: 19, up: migrateV18ToV19},
 		{version: 20, up: migrateV19ToV20},
+		{version: 21, up: migrateV20ToV21},
 	}
 }
 
@@ -1822,3 +1823,49 @@ func migrateV19ToV20(ctx context.Context, db *sql.DB) error {
 	}
 	return tx.Commit()
 }
+
+// --- v20 → v21 ---
+
+// migrateV20ToV21 restores schema-level immutability for the contents
+// table. contents is the append-only content entity: one row per BLAKE3,
+// carrying its size and origin. The id↔blake3 binding is already immutable
+// by construction (blake3 is UNIQUE), but the v13→v14 reshape dropped the
+// files_blake3_immutable trigger without installing an equivalent guard on
+// the new table, so a future bug could UPDATE a row's size_bytes/origin_*
+// in place or DELETE a row whose hash other rows still reference.
+//
+// Two triggers re-assert the guarantee the append-only contract implies:
+// any UPDATE or DELETE on a contents row aborts. The sanctioned way to
+// record different content at a path is to supersede the files row and
+// insert a new one (see Upsert), which leaves the contents row untouched.
+func migrateV20ToV21(ctx context.Context, db *sql.DB) error {
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	for _, q := range append(contentsImmutableTriggers(), `INSERT INTO schema_version (version) VALUES (21)`) {
+		if _, err := tx.ExecContext(ctx, q); err != nil {
+			return fmt.Errorf("v20→v21: %w", err)
+		}
+	}
+	return tx.Commit()
+}
+
+// contentsImmutableTriggers returns the DDL for the two triggers that make
+// the contents table append-only at the schema level: a row's size and
+// origin are fixed once written, and a row is never removed. Shared with
+// any future fresh-baseline so the guarantee survives a schema rebase.
+func contentsImmutableTriggers() []string {
+	return []string{
+		`CREATE TRIGGER contents_no_update BEFORE UPDATE ON contents
+		 BEGIN
+		     SELECT RAISE(ABORT, 'contents is append-only; supersede the files row and insert new content instead of updating');
+		 END`,
+		`CREATE TRIGGER contents_no_delete BEFORE DELETE ON contents
+		 BEGIN
+		     SELECT RAISE(ABORT, 'contents is append-only; a content row is never deleted');
+		 END`,
+	}
+}

store/runs.go

@@ -525,6 +525,10 @@ type SyncRunSpec struct {
 // sync side. Syncs to *different* destinations stay free to overlap —
 // they touch disjoint vectors.
 //
+// An in-flight offload also blocks: offload unlinks on-disk bytes the sync
+// is enumerating, and offload itself blocks on every kind, so the sync and
+// index gates name it too to keep the exclusion symmetric.
+//
 // Returns (newID, nil, nil) when the row was inserted; (0, &blocker,
 // nil) when refused — the caller is expected to render a diagnostic
 // using the blocker's id and started_at_ns. Stale rows from crashed
@@ -547,7 +551,7 @@ func (s *Store) BeginSyncRunIfClear(ctx context.Context, spec SyncRunSpec) (int6
 		WHERE status = 'running' AND volume_id = ?
 		  AND (
 		    (kind = 'sync' AND destination = ?)
-		    OR kind IN ('index', 'audit')
+		    OR kind IN ('index', 'audit', 'offload')
 		  )
 		ORDER BY id LIMIT 1
 	`, spec.VolumeID, spec.Destination)
@@ -579,14 +583,16 @@ func (s *Store) BeginSyncRunIfClear(ctx context.Context, spec SyncRunSpec) (int6
 }
 
 // BeginIndexRunIfClear atomically inserts a 'running' kind='index' or
-// kind='audit' row for volumeID iff no other index- or audit-kind run
-// is currently in flight against the same volume. Symmetric to
+// kind='audit' row for volumeID iff no index-, audit-, or offload-kind
+// run is currently in flight against the same volume. Symmetric to
 // BeginSyncRunIfClear (BEGIN IMMEDIATE + check + insert in one tx) so
 // two concurrent callers cannot both observe "no running run" and both
 // insert. Cross-kind: an in-flight 'index' blocks a new 'audit' and
 // vice versa because both walk the volume and call MarkMissing with
 // their own run-id — letting them overlap is exactly the bug this
-// guards against.
+// guards against. An in-flight offload blocks too: it unlinks bytes the
+// walk would otherwise observe and flip, and offload defers to every
+// kind, so the block is symmetric.
 //
 // A running sync does not block an index here, while a running index
 // does block a new sync (BeginSyncRunIfClear). The asymmetry is
@@ -614,7 +620,7 @@ func (s *Store) BeginIndexRunIfClear(ctx context.Context, kind string, volumeID
 	row := tx.QueryRowContext(ctx, `
 		SELECT `+runColumns+`
 		FROM runs
-		WHERE kind IN ('index', 'audit') AND status = 'running'
+		WHERE kind IN ('index', 'audit', 'offload') AND status = 'running'
 		  AND volume_id = ?
 		ORDER BY id LIMIT 1
 	`, volumeID)

store/schema.sql

@@ -1,6 +1,6 @@
 -- Generated by `go test ./store -update-schema` — DO NOT EDIT.
 --
--- Flattened snapshot of the squirrel index schema at version 20, for humans
+-- Flattened snapshot of the squirrel index schema at version 21, for humans
 -- and agents who want the current shape without replaying the migration
 -- chain in migrations.go. It is NOT used to create or migrate databases —
 -- a fresh DB is built by applyV5 plus the migration registry. The golden
@@ -17,6 +17,16 @@ CREATE TABLE contents (
 CREATE INDEX idx_contents_origin_node ON contents(origin_node_id)
 		 WHERE origin_node_id IS NOT NULL;
 
+CREATE TRIGGER contents_no_delete BEFORE DELETE ON contents
+		 BEGIN
+		     SELECT RAISE(ABORT, 'contents is append-only; a content row is never deleted');
+		 END;
+
+CREATE TRIGGER contents_no_update BEFORE UPDATE ON contents
+		 BEGIN
+		     SELECT RAISE(ABORT, 'contents is append-only; supersede the files row and insert new content instead of updating');
+		 END;
+
 CREATE TABLE destination_push_freshness (
 			volume_id      INTEGER NOT NULL REFERENCES volumes(id),
 			destination    TEXT NOT NULL,