diff --git a/internal/client/spool/events.go b/internal/client/spool/events.go index 24cbe86..3b341ea 100644 --- a/internal/client/spool/events.go +++ b/internal/client/spool/events.go @@ -86,7 +86,7 @@ func (store *Store) AcceptCommand(ctx context.Context, command Command, accepted if err != nil { return Acceptance{}, err } - _, err = tx.ExecContext(ctx, `INSERT INTO commands(issue_uuid, immutable_sha256, command_revision, phase, terminal, total_charged_bytes, closeout_remaining_bytes, accepted_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, command.IssueUUID[:], command.ImmutableSHA256[:], command.Revision, command.Phase, boolInt(command.Terminal), decision.CommandTotalCharged, decision.CloseoutRemaining, acceptedAt.UnixNano()) + _, err = tx.ExecContext(ctx, `INSERT INTO commands(issue_uuid, immutable_sha256, command_revision, phase, terminal, base_charged_bytes, total_charged_bytes, closeout_remaining_bytes, accepted_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, command.IssueUUID[:], command.ImmutableSHA256[:], command.Revision, command.Phase, boolInt(command.Terminal), charge, decision.CommandTotalCharged, decision.CloseoutRemaining, acceptedAt.UnixNano()) if err != nil { return Acceptance{}, err } diff --git a/internal/client/spool/migrations.go b/internal/client/spool/migrations.go index bfffe86..6d78f5f 100644 --- a/internal/client/spool/migrations.go +++ b/internal/client/spool/migrations.go @@ -61,6 +61,7 @@ CREATE TABLE commands ( phase INTEGER NOT NULL CHECK(phase BETWEEN 1 AND 11), terminal INTEGER NOT NULL DEFAULT 0 CHECK(terminal IN (0, 1)), output_charged_bytes INTEGER NOT NULL DEFAULT 0 CHECK(output_charged_bytes >= 0), + base_charged_bytes INTEGER NOT NULL CHECK(base_charged_bytes > 0), total_charged_bytes INTEGER NOT NULL CHECK(total_charged_bytes >= 0), closeout_remaining_bytes INTEGER NOT NULL CHECK(closeout_remaining_bytes >= 0), next_local_ordinal INTEGER NOT NULL DEFAULT 1 CHECK(next_local_ordinal > 0), diff --git a/internal/client/spool/recovery.go b/internal/client/spool/recovery.go new file mode 100644 index 0000000..6671867 --- /dev/null +++ b/internal/client/spool/recovery.go @@ -0,0 +1,137 @@ +package spool + +import ( + "bytes" + "context" + "database/sql" + "errors" + "fmt" + + "github.com/rvbox/rvbox/internal/domain" +) + +var ( + ErrIntegrityCheck = errors.New("client spool SQLite integrity check failed") + ErrQuotaCounterMismatch = errors.New("client spool quota counter mismatch") + ErrStoredPayloadChecksum = errors.New("client spool stored payload checksum mismatch") +) + +type RecoveryReport struct { + EventsChecked uint64 + ScriptsChecked uint64 +} + +// Check verifies state that SQLite's own structural check cannot express. The +// daemon runs it asynchronously during startup and exposes failure as dirty +// local health; it never invents a replacement identity or counter value. +func (store *Store) Check(ctx context.Context) (RecoveryReport, error) { + if store == nil || store.db == nil { + return RecoveryReport{}, sql.ErrConnDone + } + if err := quickCheck(ctx, store.db); err != nil { + return RecoveryReport{}, err + } + if err := checkQuotaCounters(ctx, store.db); err != nil { + return RecoveryReport{}, err + } + return checkPayloads(ctx, store.db, store.maxScriptBytes) +} + +func quickCheck(ctx context.Context, database *sql.DB) error { + rows, err := database.QueryContext(ctx, `PRAGMA quick_check(100)`) + if err != nil { + return fmt.Errorf("%w: query failed", ErrIntegrityCheck) + } + defer rows.Close() + count := 0 + for rows.Next() { + var result string + if err := rows.Scan(&result); err != nil { + return fmt.Errorf("%w: unreadable result", ErrIntegrityCheck) + } + count++ + if result != "ok" { + return fmt.Errorf("%w: %s", ErrIntegrityCheck, result) + } + } + if err := rows.Err(); err != nil || count != 1 { + return fmt.Errorf("%w: incomplete result", ErrIntegrityCheck) + } + return nil +} + +func checkQuotaCounters(ctx context.Context, database *sql.DB) error { + var invalid int + err := database.QueryRowContext(ctx, `SELECT count(*) FROM commands AS c +WHERE c.total_charged_bytes != c.base_charged_bytes + + COALESCE((SELECT sum(charged_bytes) FROM events WHERE issue_uuid = c.issue_uuid), 0) + + COALESCE((SELECT charged_bytes FROM scripts WHERE issue_uuid = c.issue_uuid), 0) +OR c.output_charged_bytes != + COALESCE((SELECT sum(charged_bytes) FROM events WHERE issue_uuid = c.issue_uuid AND output = 1), 0)`).Scan(&invalid) + if err != nil { + return err + } + var stored, computed uint64 + if err := database.QueryRowContext(ctx, `SELECT client_total_charged_bytes, COALESCE((SELECT sum(total_charged_bytes) FROM commands), 0) FROM spool_counters WHERE singleton = 1`).Scan(&stored, &computed); err != nil { + return err + } + if invalid != 0 || stored != computed { + return ErrQuotaCounterMismatch + } + return nil +} + +func checkPayloads(ctx context.Context, database *sql.DB, maxScriptBytes uint64) (RecoveryReport, error) { + var report RecoveryReport + rows, err := database.QueryContext(ctx, `SELECT issue_uuid, payload, payload_sha256 FROM events ORDER BY issue_uuid, local_ordinal`) + if err != nil { + return report, err + } + for rows.Next() { + var owner, payload, digest []byte + if err := rows.Scan(&owner, &payload, &digest); err != nil { + _ = rows.Close() + return report, err + } + computed := immutableDigest(payload) + if _, valid := copiedUUID(owner); !valid || len(digest) != 32 || !bytes.Equal(computed[:], digest) { + _ = rows.Close() + return report, ErrStoredPayloadChecksum + } + report.EventsChecked++ + } + if err := rows.Close(); err != nil { + return report, err + } + scripts, err := database.QueryContext(ctx, `SELECT issue_uuid, received_raw_bytes, stored_bytes, stored_data FROM scripts ORDER BY issue_uuid`) + if err != nil { + return report, err + } + defer scripts.Close() + for scripts.Next() { + var owner, stored []byte + var received, storedBytes uint64 + if err := scripts.Scan(&owner, &received, &storedBytes, &stored); err != nil { + return report, err + } + if _, valid := copiedUUID(owner); !valid || storedBytes != uint64(len(stored)) { + return report, ErrScriptState + } + if _, err := decompressScript(stored, received, maxScriptBytes); err != nil { + return report, err + } + report.ScriptsChecked++ + } + return report, scripts.Err() +} + +// copiedUUID is intentionally kept here to make a corrupted BLOB type obvious +// at the recovery boundary rather than relying on SQLite's dynamic conversions. +func copiedUUID(value []byte) (domain.UUID, bool) { + var result domain.UUID + if len(value) != len(result) { + return result, false + } + copy(result[:], value) + return result, validUUID(result) +} diff --git a/internal/client/spool/recovery_test.go b/internal/client/spool/recovery_test.go new file mode 100644 index 0000000..2c66e9f --- /dev/null +++ b/internal/client/spool/recovery_test.go @@ -0,0 +1,48 @@ +package spool + +import ( + "context" + "errors" + "path/filepath" + "testing" + "time" +) + +func TestCheckSpoolPayloadsAndCounters_HP_CLIENT_09(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := openTestStore(t, ctx, filepath.Join(t.TempDir(), "spool"), DefaultTombstoneLimit) + issue := testUUID(t, "019c46f1-1d02-7000-8000-000000000041") + if _, err := store.AcceptCommand(ctx, testCommand(issue, []byte("recovery")), time.Now().UTC()); err != nil { + t.Fatal(err) + } + if _, err := store.AppendEvent(ctx, issue, EventInput{Kind: 1, Compression: 1, Payload: []byte("durable event"), CreatedAt: time.Now().UTC()}); err != nil { + t.Fatal(err) + } + report, err := store.Check(ctx) + if err != nil || report.EventsChecked != 1 || report.ScriptsChecked != 0 { + t.Fatalf("initial Check = %#v, %v", report, err) + } + if _, err := store.db.ExecContext(ctx, `UPDATE events SET payload = ? WHERE issue_uuid = ?`, []byte("tampered"), issue[:]); err != nil { + t.Fatal(err) + } + if _, err := store.Check(ctx); !errors.Is(err, ErrStoredPayloadChecksum) { + t.Fatalf("tampered event Check error = %v, want ErrStoredPayloadChecksum", err) + } +} + +func TestCheckSpoolRejectsCounterDrift_BH_CLIENT_03(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := openTestStore(t, ctx, filepath.Join(t.TempDir(), "spool"), DefaultTombstoneLimit) + issue := testUUID(t, "019c46f1-1d02-7000-8000-000000000042") + if _, err := store.AcceptCommand(ctx, testCommand(issue, []byte("recovery")), time.Now().UTC()); err != nil { + t.Fatal(err) + } + if _, err := store.db.ExecContext(ctx, `UPDATE spool_counters SET client_total_charged_bytes = client_total_charged_bytes + 1`); err != nil { + t.Fatal(err) + } + if _, err := store.Check(ctx); !errors.Is(err, ErrQuotaCounterMismatch) { + t.Fatalf("counter-drift Check error = %v, want ErrQuotaCounterMismatch", err) + } +} diff --git a/test/coverage.toml b/test/coverage.toml index d6ce4ad..b901c10 100644 --- a/test/coverage.toml +++ b/test/coverage.toml @@ -233,6 +233,18 @@ layer = "unit" status = "implemented" tests = ["internal/client/spool/script_test.go:TestScriptUploadDurableExactReplayAndCommit_HP_SCRIPT_01"] +[[requirements]] +id = "HP-CLIENT-09" +layer = "unit" +status = "implemented" +tests = ["internal/client/spool/recovery_test.go:TestCheckSpoolPayloadsAndCounters_HP_CLIENT_09"] + +[[requirements]] +id = "BH-CLIENT-03" +layer = "unit" +status = "implemented" +tests = ["internal/client/spool/recovery_test.go:TestCheckSpoolRejectsCounterDrift_BH_CLIENT_03"] + [[requirements]] id = "HP-WINCTX-02" layer = "integration"