feat: verify durable client spool recovery
This commit is contained in:
@@ -86,7 +86,7 @@ func (store *Store) AcceptCommand(ctx context.Context, command Command, accepted
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return Acceptance{}, err
|
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 {
|
if err != nil {
|
||||||
return Acceptance{}, err
|
return Acceptance{}, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -61,6 +61,7 @@ CREATE TABLE commands (
|
|||||||
phase INTEGER NOT NULL CHECK(phase BETWEEN 1 AND 11),
|
phase INTEGER NOT NULL CHECK(phase BETWEEN 1 AND 11),
|
||||||
terminal INTEGER NOT NULL DEFAULT 0 CHECK(terminal IN (0, 1)),
|
terminal INTEGER NOT NULL DEFAULT 0 CHECK(terminal IN (0, 1)),
|
||||||
output_charged_bytes INTEGER NOT NULL DEFAULT 0 CHECK(output_charged_bytes >= 0),
|
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),
|
total_charged_bytes INTEGER NOT NULL CHECK(total_charged_bytes >= 0),
|
||||||
closeout_remaining_bytes INTEGER NOT NULL CHECK(closeout_remaining_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),
|
next_local_ordinal INTEGER NOT NULL DEFAULT 1 CHECK(next_local_ordinal > 0),
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -233,6 +233,18 @@ layer = "unit"
|
|||||||
status = "implemented"
|
status = "implemented"
|
||||||
tests = ["internal/client/spool/script_test.go:TestScriptUploadDurableExactReplayAndCommit_HP_SCRIPT_01"]
|
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]]
|
[[requirements]]
|
||||||
id = "HP-WINCTX-02"
|
id = "HP-WINCTX-02"
|
||||||
layer = "integration"
|
layer = "integration"
|
||||||
|
|||||||
Reference in New Issue
Block a user