feat: add durable client command spool

internal/client/spool/events.go

@@ -0,0 +1,344 @@
+package spool
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/rvbox/rvbox/internal/domain"
+)
+
+type Command struct {
+	IssueUUID       domain.UUID
+	ImmutableSHA256 [32]byte
+	Revision        uint64
+	Phase           uint32
+	Terminal        bool
+}
+
+type Acceptance struct {
+	Duplicate bool
+	Command   Command
+}
+
+type EventInput struct {
+	Kind        uint32
+	Compression uint32
+	RawBytes    uint64
+	Payload     []byte
+	CreatedAt   time.Time
+}
+
+type Event struct {
+	IssueUUID    domain.UUID
+	LocalOrdinal uint64
+	EventSeq     uint64
+	Kind         uint32
+	Compression  uint32
+	RawBytes     uint64
+	Payload      []byte
+	CreatedAt    time.Time
+}
+
+func (store *Store) AcceptCommand(ctx context.Context, command Command, acceptedAt time.Time) (Acceptance, error) {
+	if !validUUID(command.IssueUUID) || command.Revision == 0 || command.Phase == 0 || command.Phase > 11 || command.Terminal != isTerminalPhase(command.Phase) || acceptedAt.IsZero() {
+		return Acceptance{}, errors.New("invalid command acceptance")
+	}
+	tx, err := store.db.BeginTx(ctx, nil)
+	if err != nil {
+		return Acceptance{}, err
+	}
+	defer tx.Rollback()
+	existing, found, err := commandByUUID(ctx, tx, command.IssueUUID)
+	if err != nil {
+		return Acceptance{}, err
+	}
+	if found {
+		if existing.ImmutableSHA256 != command.ImmutableSHA256 {
+			return Acceptance{}, ErrCommandConflict
+		}
+		return Acceptance{Duplicate: true, Command: existing}, nil
+	}
+	var tombstoneHash []byte
+	err = tx.QueryRowContext(ctx, `SELECT immutable_sha256 FROM command_tombstones WHERE issue_uuid = ?`, command.IssueUUID[:]).Scan(&tombstoneHash)
+	if err == nil {
+		if string(tombstoneHash) != string(command.ImmutableSHA256[:]) {
+			return Acceptance{}, ErrCommandConflict
+		}
+		return Acceptance{}, ErrAlreadyExecuted
+	}
+	if err != sql.ErrNoRows {
+		return Acceptance{}, err
+	}
+	_, err = tx.ExecContext(ctx, `INSERT INTO commands(issue_uuid, immutable_sha256, command_revision, phase, terminal, accepted_at) VALUES (?, ?, ?, ?, ?, ?)`, command.IssueUUID[:], command.ImmutableSHA256[:], command.Revision, command.Phase, boolInt(command.Terminal), acceptedAt.UnixNano())
+	if err != nil {
+		return Acceptance{}, err
+	}
+	if err := tx.Commit(); err != nil {
+		return Acceptance{}, err
+	}
+	return Acceptance{Command: command}, nil
+}
+
+// AppendEvent assigns command-local order only. EventSeq remains zero until
+// AssignSendWindow makes the durable event eligible for a wire send.
+func (store *Store) AppendEvent(ctx context.Context, issueUUID domain.UUID, input EventInput) (Event, error) {
+	if !validUUID(issueUUID) || input.Kind == 0 || input.Compression < 1 || input.Compression > 2 || input.CreatedAt.IsZero() {
+		return Event{}, errors.New("invalid client event")
+	}
+	tx, err := store.db.BeginTx(ctx, nil)
+	if err != nil {
+		return Event{}, err
+	}
+	defer tx.Rollback()
+	var ordinal uint64
+	err = tx.QueryRowContext(ctx, `UPDATE commands SET next_local_ordinal = next_local_ordinal + 1 WHERE issue_uuid = ? RETURNING next_local_ordinal - 1`, issueUUID[:]).Scan(&ordinal)
+	if err == sql.ErrNoRows {
+		return Event{}, ErrUnknownCommand
+	}
+	if err != nil {
+		return Event{}, err
+	}
+	digest := immutableDigest(input.Payload)
+	_, err = tx.ExecContext(ctx, `INSERT INTO events(issue_uuid, local_ordinal, event_kind, compression, raw_bytes, payload, payload_sha256, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, issueUUID[:], ordinal, input.Kind, input.Compression, input.RawBytes, input.Payload, digest[:], input.CreatedAt.UnixNano())
+	if err != nil {
+		return Event{}, err
+	}
+	if err := tx.Commit(); err != nil {
+		return Event{}, err
+	}
+	return Event{IssueUUID: issueUUID, LocalOrdinal: ordinal, Kind: input.Kind, Compression: input.Compression, RawBytes: input.RawBytes, Payload: append([]byte(nil), input.Payload...), CreatedAt: input.CreatedAt}, nil
+}
+
+// AssignSendWindow atomically gives previously-unassigned events contiguous
+// per-command wire sequences. Once assigned, they are immutable until Ack.
+func (store *Store) AssignSendWindow(ctx context.Context, issueUUID domain.UUID, maximumEvents, maximumStoredBytes uint64) ([]Event, error) {
+	if maximumEvents == 0 || maximumStoredBytes == 0 {
+		return nil, nil
+	}
+	tx, err := store.db.BeginTx(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+	var next uint64
+	err = tx.QueryRowContext(ctx, `SELECT next_event_seq FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&next)
+	if err == sql.ErrNoRows {
+		return nil, ErrUnknownCommand
+	}
+	if err != nil {
+		return nil, err
+	}
+	rows, err := tx.QueryContext(ctx, `SELECT local_ordinal, length(payload) FROM events WHERE issue_uuid = ? AND event_seq IS NULL ORDER BY local_ordinal LIMIT ?`, issueUUID[:], maximumEvents)
+	if err != nil {
+		return nil, err
+	}
+	var ordinals []uint64
+	for rows.Next() {
+		var ordinal, storedBytes uint64
+		if err := rows.Scan(&ordinal, &storedBytes); err != nil {
+			_ = rows.Close()
+			return nil, err
+		}
+		if storedBytes > maximumStoredBytes {
+			break
+		}
+		maximumStoredBytes -= storedBytes
+		ordinals = append(ordinals, ordinal)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	for _, ordinal := range ordinals {
+		if _, err := tx.ExecContext(ctx, `UPDATE events SET event_seq = ? WHERE issue_uuid = ? AND local_ordinal = ? AND event_seq IS NULL`, next, issueUUID[:], ordinal); err != nil {
+			return nil, err
+		}
+		next++
+	}
+	if len(ordinals) > 0 {
+		if _, err := tx.ExecContext(ctx, `UPDATE commands SET next_event_seq = ? WHERE issue_uuid = ?`, next, issueUUID[:]); err != nil {
+			return nil, err
+		}
+	}
+	events, err := eventsBySequence(ctx, tx, issueUUID, 0)
+	if err != nil {
+		return nil, err
+	}
+	if err := tx.Commit(); err != nil {
+		return nil, err
+	}
+	return events, nil
+}
+
+func (store *Store) PendingEvents(ctx context.Context, issueUUID domain.UUID) ([]Event, error) {
+	return eventsBySequence(ctx, store.db, issueUUID, 0)
+}
+
+// Ack applies a cumulative server acknowledgement. Acknowledging an event that
+// was never assigned is a protocol/storage boundary violation, never a hint to
+// skip data.
+func (store *Store) Ack(ctx context.Context, issueUUID domain.UUID, eventSeq uint64) error {
+	tx, err := store.db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	var lastAck, next uint64
+	err = tx.QueryRowContext(ctx, `SELECT last_server_ack, next_event_seq FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&lastAck, &next)
+	if err == sql.ErrNoRows {
+		return ErrUnknownCommand
+	}
+	if err != nil {
+		return err
+	}
+	if eventSeq <= lastAck {
+		return tx.Commit()
+	}
+	if eventSeq >= next {
+		return ErrInvalidEventAck
+	}
+	if _, err := tx.ExecContext(ctx, `DELETE FROM events WHERE issue_uuid = ? AND event_seq <= ?`, issueUUID[:], eventSeq); err != nil {
+		return err
+	}
+	if _, err := tx.ExecContext(ctx, `UPDATE commands SET last_server_ack = ? WHERE issue_uuid = ?`, eventSeq, issueUUID[:]); err != nil {
+		return err
+	}
+	return tx.Commit()
+}
+
+func (store *Store) MarkTerminal(ctx context.Context, issueUUID domain.UUID, phase uint32) error {
+	if phase < 5 || phase > 11 {
+		return errors.New("invalid terminal command phase")
+	}
+	result, err := store.db.ExecContext(ctx, `UPDATE commands SET phase = ?, terminal = 1 WHERE issue_uuid = ?`, phase, issueUUID[:])
+	if err != nil {
+		return err
+	}
+	count, err := result.RowsAffected()
+	if err != nil {
+		return err
+	}
+	if count == 0 {
+		return ErrUnknownCommand
+	}
+	return nil
+}
+
+// CleanupTerminal moves a fully acknowledged terminal command into the compact
+// tombstone ledger and removes all command-owned spool data in one transaction.
+func (store *Store) CleanupTerminal(ctx context.Context, issueUUID domain.UUID, acknowledgedAt time.Time) error {
+	if acknowledgedAt.IsZero() {
+		return errors.New("terminal acknowledgement time is required")
+	}
+	tx, err := store.db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	var hash []byte
+	var terminal int
+	err = tx.QueryRowContext(ctx, `SELECT immutable_sha256, terminal FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&hash, &terminal)
+	if err == sql.ErrNoRows {
+		return ErrUnknownCommand
+	}
+	if err != nil {
+		return err
+	}
+	if terminal == 0 {
+		return errors.New("command is not terminal")
+	}
+	var pending uint64
+	if err := tx.QueryRowContext(ctx, `SELECT count(*) FROM events WHERE issue_uuid = ?`, issueUUID[:]).Scan(&pending); err != nil {
+		return err
+	}
+	if pending != 0 {
+		return ErrEventsPending
+	}
+	if _, err := tx.ExecContext(ctx, `INSERT INTO command_tombstones(issue_uuid, immutable_sha256, acknowledged_at) VALUES (?, ?, ?)`, issueUUID[:], hash, acknowledgedAt.UnixNano()); err != nil {
+		return err
+	}
+	if _, err := tx.ExecContext(ctx, `DELETE FROM commands WHERE issue_uuid = ?`, issueUUID[:]); err != nil {
+		return err
+	}
+	trim, err := tombstonesToTrim(ctx, tx, store.tombstoneLimit)
+	if err != nil {
+		return err
+	}
+	if _, err := tx.ExecContext(ctx, `DELETE FROM command_tombstones WHERE issue_uuid IN (SELECT issue_uuid FROM command_tombstones ORDER BY acknowledged_at, issue_uuid LIMIT ?)`, trim); err != nil {
+		return err
+	}
+	return tx.Commit()
+}
+
+type queryer interface {
+	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
+}
+
+func eventsBySequence(ctx context.Context, query queryer, issueUUID domain.UUID, after uint64) ([]Event, error) {
+	rows, err := query.QueryContext(ctx, `SELECT local_ordinal, event_seq, event_kind, compression, raw_bytes, payload, created_at FROM events WHERE issue_uuid = ? AND event_seq IS NOT NULL AND event_seq > ? ORDER BY event_seq`, issueUUID[:], after)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var events []Event
+	for rows.Next() {
+		var event Event
+		var created int64
+		event.IssueUUID = issueUUID
+		if err := rows.Scan(&event.LocalOrdinal, &event.EventSeq, &event.Kind, &event.Compression, &event.RawBytes, &event.Payload, &created); err != nil {
+			return nil, err
+		}
+		event.CreatedAt = time.Unix(0, created).UTC()
+		events = append(events, event)
+	}
+	return events, rows.Err()
+}
+
+func commandByUUID(ctx context.Context, query interface {
+	QueryRowContext(context.Context, string, ...any) *sql.Row
+}, issueUUID domain.UUID) (Command, bool, error) {
+	var command Command
+	var hash []byte
+	var terminal int
+	err := query.QueryRowContext(ctx, `SELECT immutable_sha256, command_revision, phase, terminal FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&hash, &command.Revision, &command.Phase, &terminal)
+	if err == sql.ErrNoRows {
+		return Command{}, false, nil
+	}
+	if err != nil {
+		return Command{}, false, err
+	}
+	if len(hash) != len(command.ImmutableSHA256) {
+		return Command{}, false, fmt.Errorf("invalid stored immutable command hash")
+	}
+	copy(command.ImmutableSHA256[:], hash)
+	command.IssueUUID = issueUUID
+	command.Terminal = terminal != 0
+	return command, true, nil
+}
+
+func tombstonesToTrim(ctx context.Context, tx *sql.Tx, limit uint64) (uint64, error) {
+	var count uint64
+	if err := tx.QueryRowContext(ctx, `SELECT count(*) FROM command_tombstones`).Scan(&count); err != nil {
+		return 0, err
+	}
+	if count <= limit {
+		return 0, nil
+	}
+	return count - limit, nil
+}
+
+func boolInt(value bool) int {
+	if value {
+		return 1
+	}
+	return 0
+}
+
+func validUUID(value domain.UUID) bool {
+	_, err := domain.ParseUUIDv7(value.String())
+	return err == nil
+}
+
+func isTerminalPhase(phase uint32) bool { return phase >= 5 && phase <= 11 }

internal/client/spool/identity.go

@@ -0,0 +1,75 @@
+package spool
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/rvbox/rvbox/internal/domain"
+)
+
+func loadOrCreateIdentity(path string, generate func() (domain.UUID, error)) (domain.UUID, error) {
+	data, err := os.ReadFile(path)
+	if err == nil {
+		return parseIdentity(data)
+	}
+	if !errors.Is(err, os.ErrNotExist) {
+		return domain.UUID{}, fmt.Errorf("read client instance identity: %w", err)
+	}
+	identity, err := generate()
+	if err != nil {
+		return domain.UUID{}, fmt.Errorf("generate client instance identity: %w", err)
+	}
+	if err := writeIdentityAtomically(path, []byte(identity.String()+"\n")); err != nil {
+		return domain.UUID{}, err
+	}
+	return identity, nil
+}
+
+func parseIdentity(data []byte) (domain.UUID, error) {
+	if len(data) != 37 || data[36] != '\n' {
+		return domain.UUID{}, ErrIdentityCorrupt
+	}
+	identity, err := domain.ParseUUIDv7(string(data[:36]))
+	if err != nil {
+		return domain.UUID{}, fmt.Errorf("%w: %v", ErrIdentityCorrupt, err)
+	}
+	return identity, nil
+}
+
+func writeIdentityAtomically(path string, data []byte) (result error) {
+	directory := filepath.Dir(path)
+	temporary, err := os.CreateTemp(directory, ".client-instance-id-*")
+	if err != nil {
+		return fmt.Errorf("create client identity temporary file: %w", err)
+	}
+	temporaryPath := temporary.Name()
+	defer func() {
+		if result != nil {
+			_ = os.Remove(temporaryPath)
+		}
+	}()
+	if err := temporary.Chmod(0o600); err != nil {
+		_ = temporary.Close()
+		return fmt.Errorf("secure client identity temporary file: %w", err)
+	}
+	if _, err := temporary.Write(data); err != nil {
+		_ = temporary.Close()
+		return fmt.Errorf("write client identity: %w", err)
+	}
+	if err := temporary.Sync(); err != nil {
+		_ = temporary.Close()
+		return fmt.Errorf("sync client identity: %w", err)
+	}
+	if err := temporary.Close(); err != nil {
+		return fmt.Errorf("close client identity: %w", err)
+	}
+	if err := replaceIdentityFile(temporaryPath, path); err != nil {
+		return fmt.Errorf("install client identity: %w", err)
+	}
+	if err := syncDirectory(directory); err != nil {
+		return fmt.Errorf("sync client identity directory: %w", err)
+	}
+	return nil
+}

internal/client/spool/lock_unix.go

@@ -0,0 +1,47 @@
+//go:build !windows
+
+package spool
+
+import (
+	"errors"
+	"os"
+	"syscall"
+)
+
+func acquireInstanceLock(path string) (func() error, error) {
+	if err := ensurePrivateFile(path); err != nil {
+		return nil, err
+	}
+	file, err := os.OpenFile(path, os.O_RDWR, 0)
+	if err != nil {
+		return nil, err
+	}
+	if err := syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
+		_ = file.Close()
+		if errors.Is(err, syscall.EWOULDBLOCK) {
+			return nil, ErrAlreadyOpen
+		}
+		return nil, err
+	}
+	return func() error {
+		unlockErr := syscall.Flock(int(file.Fd()), syscall.LOCK_UN)
+		closeErr := file.Close()
+		if unlockErr != nil {
+			return unlockErr
+		}
+		return closeErr
+	}, nil
+}
+
+func syncDirectory(path string) error {
+	directory, err := os.Open(path)
+	if err != nil {
+		return err
+	}
+	defer directory.Close()
+	return directory.Sync()
+}
+
+func replaceIdentityFile(temporaryPath, targetPath string) error {
+	return os.Rename(temporaryPath, targetPath)
+}

internal/client/spool/lock_windows.go

@@ -0,0 +1,58 @@
+//go:build windows
+
+package spool
+
+import (
+	"os"
+
+	"golang.org/x/sys/windows"
+)
+
+func acquireInstanceLock(path string) (func() error, error) {
+	if err := ensurePrivateFile(path); err != nil {
+		return nil, err
+	}
+	file, err := os.OpenFile(path, os.O_RDWR, 0)
+	if err != nil {
+		return nil, err
+	}
+	var overlapped windows.Overlapped
+	err = windows.LockFileEx(windows.Handle(file.Fd()), windows.LOCKFILE_EXCLUSIVE_LOCK|windows.LOCKFILE_FAIL_IMMEDIATELY, 0, 1, 0, &overlapped)
+	if err != nil {
+		_ = file.Close()
+		return nil, ErrAlreadyOpen
+	}
+	return func() error {
+		unlockErr := windows.UnlockFileEx(windows.Handle(file.Fd()), 0, 1, 0, &overlapped)
+		closeErr := file.Close()
+		if unlockErr != nil {
+			return unlockErr
+		}
+		return closeErr
+	}, nil
+}
+
+func syncDirectory(path string) error {
+	value, err := windows.UTF16PtrFromString(path)
+	if err != nil {
+		return err
+	}
+	handle, err := windows.CreateFile(value, windows.GENERIC_READ, windows.FILE_SHARE_READ|windows.FILE_SHARE_WRITE|windows.FILE_SHARE_DELETE, nil, windows.OPEN_EXISTING, windows.FILE_FLAG_BACKUP_SEMANTICS, 0)
+	if err != nil {
+		return err
+	}
+	defer windows.CloseHandle(handle)
+	return windows.FlushFileBuffers(handle)
+}
+
+func replaceIdentityFile(temporaryPath, targetPath string) error {
+	from, err := windows.UTF16PtrFromString(temporaryPath)
+	if err != nil {
+		return err
+	}
+	to, err := windows.UTF16PtrFromString(targetPath)
+	if err != nil {
+		return err
+	}
+	return windows.MoveFileEx(from, to, windows.MOVEFILE_REPLACE_EXISTING|windows.MOVEFILE_WRITE_THROUGH)
+}

internal/client/spool/migrations.go

@@ -0,0 +1,88 @@
+package spool
+
+import (
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/hex"
+	"fmt"
+)
+
+type migration struct {
+	version uint32
+	sql     string
+}
+
+var migrations = []migration{{version: 1, sql: schemaV1}}
+
+func applyMigrations(ctx context.Context, db *sql.DB) error {
+	if _, err := db.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations (
+version INTEGER PRIMARY KEY CHECK(version > 0), checksum TEXT NOT NULL, applied_at INTEGER NOT NULL
+) STRICT`); err != nil {
+		return fmt.Errorf("create client spool migration table: %w", err)
+	}
+	for _, current := range migrations {
+		checksumBytes := sha256.Sum256([]byte(current.sql))
+		checksum := hex.EncodeToString(checksumBytes[:])
+		var stored string
+		err := db.QueryRowContext(ctx, `SELECT checksum FROM schema_migrations WHERE version = ?`, current.version).Scan(&stored)
+		if err == nil {
+			if stored != checksum {
+				return fmt.Errorf("client spool migration %d checksum mismatch", current.version)
+			}
+			continue
+		}
+		if err != sql.ErrNoRows {
+			return err
+		}
+		tx, err := db.BeginTx(ctx, nil)
+		if err != nil {
+			return err
+		}
+		if _, err = tx.ExecContext(ctx, current.sql); err == nil {
+			_, err = tx.ExecContext(ctx, `INSERT INTO schema_migrations(version, checksum, applied_at) VALUES (?, ?, unixepoch())`, current.version, checksum)
+		}
+		if err != nil {
+			_ = tx.Rollback()
+			return fmt.Errorf("apply client spool migration %d: %w", current.version, err)
+		}
+		if err := tx.Commit(); err != nil {
+			return fmt.Errorf("commit client spool migration %d: %w", current.version, err)
+		}
+	}
+	return nil
+}
+
+const schemaV1 = `
+CREATE TABLE commands (
+  issue_uuid BLOB PRIMARY KEY CHECK(length(issue_uuid) = 16),
+  immutable_sha256 BLOB NOT NULL CHECK(length(immutable_sha256) = 32),
+  command_revision INTEGER NOT NULL CHECK(command_revision > 0),
+  phase INTEGER NOT NULL CHECK(phase BETWEEN 1 AND 11),
+  terminal INTEGER NOT NULL DEFAULT 0 CHECK(terminal IN (0, 1)),
+  next_local_ordinal INTEGER NOT NULL DEFAULT 1 CHECK(next_local_ordinal > 0),
+  next_event_seq INTEGER NOT NULL DEFAULT 1 CHECK(next_event_seq > 0),
+  last_server_ack INTEGER NOT NULL DEFAULT 0 CHECK(last_server_ack >= 0),
+  accepted_at INTEGER NOT NULL
+) STRICT;
+CREATE TABLE events (
+  issue_uuid BLOB NOT NULL REFERENCES commands(issue_uuid) ON DELETE CASCADE,
+  local_ordinal INTEGER NOT NULL CHECK(local_ordinal > 0),
+  event_seq INTEGER CHECK(event_seq IS NULL OR event_seq > 0),
+  event_kind INTEGER NOT NULL CHECK(event_kind > 0),
+  compression INTEGER NOT NULL CHECK(compression IN (1, 2)),
+  raw_bytes INTEGER NOT NULL CHECK(raw_bytes >= 0),
+  payload BLOB NOT NULL,
+  payload_sha256 BLOB NOT NULL CHECK(length(payload_sha256) = 32),
+  created_at INTEGER NOT NULL,
+  PRIMARY KEY(issue_uuid, local_ordinal),
+  UNIQUE(issue_uuid, event_seq)
+) STRICT, WITHOUT ROWID;
+CREATE INDEX events_send_window ON events(issue_uuid, event_seq, local_ordinal);
+CREATE TABLE command_tombstones (
+  issue_uuid BLOB PRIMARY KEY CHECK(length(issue_uuid) = 16),
+  immutable_sha256 BLOB NOT NULL CHECK(length(immutable_sha256) = 32),
+  acknowledged_at INTEGER NOT NULL
+) STRICT;
+CREATE INDEX tombstones_fifo ON command_tombstones(acknowledged_at, issue_uuid);
+`

internal/client/spool/spool.go

@@ -0,0 +1,170 @@
+// Package spool owns the durable, command-local state retained by a client
+// daemon between network sessions and process restarts.
+package spool
+
+import (
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"errors"
+	"fmt"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/rvbox/rvbox/internal/domain"
+
+	_ "modernc.org/sqlite"
+)
+
+var (
+	ErrUnsafeDataDirectory = errors.New("unsafe client spool directory")
+	ErrAlreadyOpen         = errors.New("client spool directory is already locked")
+	ErrIdentityCorrupt     = errors.New("client instance identity is corrupt")
+	ErrCommandConflict     = errors.New("command UUID has different immutable content")
+	ErrAlreadyExecuted     = errors.New("command UUID was already executed")
+	ErrUnknownCommand      = errors.New("unknown client command")
+	ErrInvalidEventAck     = errors.New("event acknowledgement is beyond assigned sequence")
+	ErrEventsPending       = errors.New("command has events pending server acknowledgement")
+)
+
+const DefaultTombstoneLimit uint64 = 1_000_000
+
+type Options struct {
+	DataDir        string
+	BusyTimeout    time.Duration
+	TombstoneLimit uint64
+}
+
+type Store struct {
+	db             *sql.DB
+	unlock         func() error
+	dataDir        string
+	identity       domain.UUID
+	tombstoneLimit uint64
+	mu             sync.Mutex
+}
+
+// Open recovers an existing spool or creates an empty one. The caller must
+// surface ErrIdentityCorrupt as dirty health; it is deliberately never healed
+// by assigning a new client identity.
+func Open(ctx context.Context, options Options) (*Store, error) {
+	if !filepath.IsAbs(options.DataDir) || filepath.Clean(options.DataDir) == string(filepath.Separator) {
+		return nil, ErrUnsafeDataDirectory
+	}
+	if options.BusyTimeout <= 0 {
+		return nil, errors.New("busy timeout must be positive")
+	}
+	if options.TombstoneLimit == 0 {
+		options.TombstoneLimit = DefaultTombstoneLimit
+	}
+	if err := ensurePrivateDirectory(options.DataDir); err != nil {
+		return nil, err
+	}
+	unlock, err := acquireInstanceLock(filepath.Join(options.DataDir, "spool.lock"))
+	if err != nil {
+		return nil, err
+	}
+	identity, err := loadOrCreateIdentity(filepath.Join(options.DataDir, "client-instance-id"), domain.NewUUIDv7)
+	if err != nil {
+		_ = unlock()
+		return nil, err
+	}
+	databasePath := filepath.Join(options.DataDir, "spool.db")
+	if err := ensurePrivateFile(databasePath); err != nil {
+		_ = unlock()
+		return nil, err
+	}
+	query := url.Values{}
+	query.Add("_defensive", "1")
+	query.Add("_pragma", "journal_mode(WAL)")
+	query.Add("_pragma", "foreign_keys(ON)")
+	query.Add("_pragma", "synchronous(FULL)")
+	query.Add("_pragma", "busy_timeout("+strconv.FormatInt(options.BusyTimeout.Milliseconds(), 10)+")")
+	databaseURL := &url.URL{Scheme: "file", Path: filepath.ToSlash(databasePath), RawQuery: query.Encode()}
+	db, err := sql.Open("sqlite", databaseURL.String())
+	if err != nil {
+		_ = unlock()
+		return nil, err
+	}
+	db.SetMaxOpenConns(1)
+	db.SetMaxIdleConns(1)
+	store := &Store{db: db, unlock: unlock, dataDir: options.DataDir, identity: identity, tombstoneLimit: options.TombstoneLimit}
+	if err := db.PingContext(ctx); err != nil {
+		_ = store.Close()
+		return nil, fmt.Errorf("open client spool SQLite: %w", err)
+	}
+	if err := applyMigrations(ctx, db); err != nil {
+		_ = store.Close()
+		return nil, err
+	}
+	return store, nil
+}
+
+func (store *Store) ClientInstanceID() domain.UUID { return store.identity }
+
+func (store *Store) Close() error {
+	if store == nil {
+		return nil
+	}
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	var result error
+	if store.db != nil {
+		result = store.db.Close()
+		store.db = nil
+	}
+	if store.unlock != nil {
+		if err := store.unlock(); result == nil {
+			result = err
+		}
+		store.unlock = nil
+	}
+	return result
+}
+
+func ensurePrivateFile(path string) error {
+	file, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_RDWR, 0o600)
+	if err == nil {
+		return file.Close()
+	}
+	if !errors.Is(err, os.ErrExist) {
+		return err
+	}
+	info, err := os.Lstat(path)
+	if err != nil {
+		return err
+	}
+	if !info.Mode().IsRegular() || info.Mode()&os.ModeSymlink != 0 {
+		return fmt.Errorf("%w: %s is not a regular file", ErrUnsafeDataDirectory, path)
+	}
+	if info.Mode().Perm()&0o077 != 0 {
+		return fmt.Errorf("%w: %s permissions %04o expose private state", ErrUnsafeDataDirectory, path, info.Mode().Perm())
+	}
+	return nil
+}
+
+func ensurePrivateDirectory(path string) error {
+	info, err := os.Lstat(path)
+	if os.IsNotExist(err) {
+		if err := os.MkdirAll(path, 0o700); err != nil {
+			return err
+		}
+		info, err = os.Lstat(path)
+	}
+	if err != nil {
+		return err
+	}
+	if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
+		return fmt.Errorf("%w: %s is not a real directory", ErrUnsafeDataDirectory, path)
+	}
+	if info.Mode().Perm()&0o077 != 0 {
+		return fmt.Errorf("%w: %s permissions %04o expose private state", ErrUnsafeDataDirectory, path, info.Mode().Perm())
+	}
+	return nil
+}
+
+func immutableDigest(payload []byte) [32]byte { return sha256.Sum256(payload) }

internal/client/spool/spool_test.go

@@ -0,0 +1,182 @@
+package spool
+
+import (
+	"context"
+	"crypto/sha256"
+	"errors"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/rvbox/rvbox/internal/domain"
+)
+
+func TestStoreDurableIdentityAndSingleInstance_HP_CLIENT_01(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	directory := filepath.Join(t.TempDir(), "spool")
+	store := openTestStore(t, ctx, directory, DefaultTombstoneLimit)
+	identity := store.ClientInstanceID()
+	if identity == (domain.UUID{}) {
+		t.Fatal("client instance ID is zero")
+	}
+	if _, err := Open(ctx, Options{DataDir: directory, BusyTimeout: time.Second}); !errors.Is(err, ErrAlreadyOpen) {
+		t.Fatalf("second Open error = %v, want ErrAlreadyOpen", err)
+	}
+	if err := store.Close(); err != nil {
+		t.Fatal(err)
+	}
+	reopened := openTestStore(t, ctx, directory, DefaultTombstoneLimit)
+	if got := reopened.ClientInstanceID(); got != identity {
+		t.Fatalf("reopened client instance ID = %s, want %s", got, identity)
+	}
+}
+
+func TestStoreRejectsCorruptIdentity_BH_CLIENT_01(t *testing.T) {
+	t.Parallel()
+	directory := t.TempDir()
+	if err := os.Chmod(directory, 0o700); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(directory, "client-instance-id"), []byte("not-an-id\n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	_, err := Open(context.Background(), Options{DataDir: directory, BusyTimeout: time.Second})
+	if !errors.Is(err, ErrIdentityCorrupt) {
+		t.Fatalf("Open corrupt identity error = %v, want ErrIdentityCorrupt", err)
+	}
+	data, err := os.ReadFile(filepath.Join(directory, "client-instance-id"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(data) != "not-an-id\n" {
+		t.Fatalf("corrupt identity was rewritten as %q", data)
+	}
+}
+
+func TestSpoolAcceptanceSequencingAndAck_HP_CLIENT_07(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	directory := filepath.Join(t.TempDir(), "spool")
+	store := openTestStore(t, ctx, directory, DefaultTombstoneLimit)
+	issue := testUUID(t, "019c46f1-1d02-7000-8000-000000000001")
+	command := testCommand(issue, []byte("immutable command"))
+	acceptedAt := time.Date(2026, time.September, 6, 12, 0, 0, 0, time.UTC)
+	accepted, err := store.AcceptCommand(ctx, command, acceptedAt)
+	if err != nil || accepted.Duplicate {
+		t.Fatalf("first acceptance = %#v, %v", accepted, err)
+	}
+	duplicate, err := store.AcceptCommand(ctx, command, acceptedAt.Add(time.Second))
+	if err != nil || !duplicate.Duplicate || duplicate.Command.Revision != command.Revision {
+		t.Fatalf("duplicate acceptance = %#v, %v", duplicate, err)
+	}
+	first, err := store.AppendEvent(ctx, issue, EventInput{Kind: 1, Compression: 1, RawBytes: 5, Payload: []byte("first"), CreatedAt: acceptedAt})
+	if err != nil {
+		t.Fatal(err)
+	}
+	second, err := store.AppendEvent(ctx, issue, EventInput{Kind: 2, Compression: 1, RawBytes: 6, Payload: []byte("second"), CreatedAt: acceptedAt.Add(time.Nanosecond)})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if first.LocalOrdinal != 1 || first.EventSeq != 0 || second.LocalOrdinal != 2 || second.EventSeq != 0 {
+		t.Fatalf("unassigned events = %#v, %#v", first, second)
+	}
+	assigned, err := store.AssignSendWindow(ctx, issue, 1, 1<<20)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(assigned) != 1 || assigned[0].LocalOrdinal != 1 || assigned[0].EventSeq != 1 {
+		t.Fatalf("first send window = %#v", assigned)
+	}
+	assigned, err = store.AssignSendWindow(ctx, issue, 1, 1<<20)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(assigned) != 2 || assigned[1].LocalOrdinal != 2 || assigned[1].EventSeq != 2 {
+		t.Fatalf("replayed send window = %#v", assigned)
+	}
+	if err := store.Ack(ctx, issue, 1); err != nil {
+		t.Fatal(err)
+	}
+	pending, err := store.PendingEvents(ctx, issue)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(pending) != 1 || pending[0].EventSeq != 2 || string(pending[0].Payload) != "second" {
+		t.Fatalf("pending after cumulative ack = %#v", pending)
+	}
+	if err := store.Ack(ctx, issue, 3); !errors.Is(err, ErrInvalidEventAck) {
+		t.Fatalf("ack beyond send window error = %v, want ErrInvalidEventAck", err)
+	}
+	if err := store.Close(); err != nil {
+		t.Fatal(err)
+	}
+	store = openTestStore(t, ctx, directory, DefaultTombstoneLimit)
+	pending, err = store.PendingEvents(ctx, issue)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(pending) != 1 || pending[0].EventSeq != 2 {
+		t.Fatalf("pending after reopen = %#v", pending)
+	}
+}
+
+func TestTerminalCleanupTombstonesAndConflicts_BH_CLIENT_02(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	store := openTestStore(t, ctx, filepath.Join(t.TempDir(), "spool"), 1)
+	first := testCommand(testUUID(t, "019c46f1-1d02-7000-8000-000000000011"), []byte("first"))
+	second := testCommand(testUUID(t, "019c46f1-1d02-7000-8000-000000000012"), []byte("second"))
+	now := time.Date(2026, time.September, 6, 12, 0, 0, 0, time.UTC)
+	for _, command := range []Command{first, second} {
+		if _, err := store.AcceptCommand(ctx, command, now); err != nil {
+			t.Fatal(err)
+		}
+		if err := store.MarkTerminal(ctx, command.IssueUUID, 5); err != nil {
+			t.Fatal(err)
+		}
+		if err := store.CleanupTerminal(ctx, command.IssueUUID, now); err != nil {
+			t.Fatal(err)
+		}
+		now = now.Add(time.Nanosecond)
+	}
+	if _, err := store.AcceptCommand(ctx, second, now); !errors.Is(err, ErrAlreadyExecuted) {
+		t.Fatalf("exact tombstone replay error = %v, want ErrAlreadyExecuted", err)
+	}
+	conflict := second
+	conflict.ImmutableSHA256 = sha256.Sum256([]byte("changed"))
+	if _, err := store.AcceptCommand(ctx, conflict, now); !errors.Is(err, ErrCommandConflict) {
+		t.Fatalf("conflicting tombstone replay error = %v, want ErrCommandConflict", err)
+	}
+	if _, err := store.AcceptCommand(ctx, first, now); err != nil {
+		t.Fatalf("oldest tombstone was not rotated: %v", err)
+	}
+}
+
+func openTestStore(t *testing.T, ctx context.Context, directory string, tombstoneLimit uint64) *Store {
+	t.Helper()
+	store, err := Open(ctx, Options{DataDir: directory, BusyTimeout: time.Second, TombstoneLimit: tombstoneLimit})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if err := store.Close(); err != nil {
+			t.Error(err)
+		}
+	})
+	return store
+}
+
+func testUUID(t *testing.T, value string) domain.UUID {
+	t.Helper()
+	parsed, err := domain.ParseUUIDv7(value)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return parsed
+}
+
+func testCommand(issue domain.UUID, immutable []byte) Command {
+	return Command{IssueUUID: issue, ImmutableSHA256: sha256.Sum256(immutable), Revision: 1, Phase: 1}
+}

test/coverage.toml

@@ -105,7 +105,28 @@ tests = [
 id = "HP-CLIENT-07"
 layer = "unit"
 status = "implemented"
-tests = ["internal/domain/sequence_test.go:TestEventSequenceAndDuplicateEquivalence_HP_CLIENT_07"]
+tests = [
+  "internal/domain/sequence_test.go:TestEventSequenceAndDuplicateEquivalence_HP_CLIENT_07",
+  "internal/client/spool/spool_test.go:TestSpoolAcceptanceSequencingAndAck_HP_CLIENT_07",
+]
+
+[[requirements]]
+id = "HP-CLIENT-01"
+layer = "unit"
+status = "implemented"
+tests = ["internal/client/spool/spool_test.go:TestStoreDurableIdentityAndSingleInstance_HP_CLIENT_01"]
+
+[[requirements]]
+id = "BH-CLIENT-01"
+layer = "unit"
+status = "implemented"
+tests = ["internal/client/spool/spool_test.go:TestStoreRejectsCorruptIdentity_BH_CLIENT_01"]
+
+[[requirements]]
+id = "BH-CLIENT-02"
+layer = "unit"
+status = "implemented"
+tests = ["internal/client/spool/spool_test.go:TestTerminalCleanupTombstonesAndConflicts_BH_CLIENT_02"]
 
 [[requirements]]
 id = "HP-WINCTX-01"