diff --git a/internal/client/spool/events.go b/internal/client/spool/events.go new file mode 100644 index 0000000..4194c26 --- /dev/null +++ b/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 } diff --git a/internal/client/spool/identity.go b/internal/client/spool/identity.go new file mode 100644 index 0000000..eb0945d --- /dev/null +++ b/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 +} diff --git a/internal/client/spool/lock_unix.go b/internal/client/spool/lock_unix.go new file mode 100644 index 0000000..a5b83fe --- /dev/null +++ b/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) +} diff --git a/internal/client/spool/lock_windows.go b/internal/client/spool/lock_windows.go new file mode 100644 index 0000000..d1015cd --- /dev/null +++ b/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) +} diff --git a/internal/client/spool/migrations.go b/internal/client/spool/migrations.go new file mode 100644 index 0000000..ed927de --- /dev/null +++ b/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); +` diff --git a/internal/client/spool/spool.go b/internal/client/spool/spool.go new file mode 100644 index 0000000..355be7f --- /dev/null +++ b/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) } diff --git a/internal/client/spool/spool_test.go b/internal/client/spool/spool_test.go new file mode 100644 index 0000000..b025d42 --- /dev/null +++ b/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} +} diff --git a/test/coverage.toml b/test/coverage.toml index c629899..fcc9293 100644 --- a/test/coverage.toml +++ b/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"