feat: persist events across commit boundaries

2026-08-31 09:35:20 +00:00
parent 43dcd56313
commit 583b539f86
7 changed files with 722 additions and 14 deletions
@@ -0,0 +1,238 @@
+package store
+
+import (
+	"context"
+	"crypto/subtle"
+	"database/sql"
+	"errors"
+	"fmt"
+	"io/fs"
+	"math"
+	"path/filepath"
+)
+
+const (
+	FaultAfterSegmentSyncBeforeMetadata = "after_segment_sync_before_metadata"
+	FaultAfterMetadataCommit            = "after_metadata_commit"
+)
+
+var (
+	ErrStoreClosed      = errors.New("store is closed")
+	ErrCommandNotFound  = errors.New("command not found")
+	ErrEventConflict    = errors.New("event sequence already has different immutable content")
+	ErrEventSequenceGap = errors.New("event sequence is not the next expected value")
+)
+
+type FaultInjector interface {
+	Checkpoint(name string) error
+}
+
+type FaultInjectorFunc func(name string) error
+
+func (function FaultInjectorFunc) Checkpoint(name string) error { return function(name) }
+
+type EventAppend struct {
+	IssueUUID        [16]byte
+	EventSeq         uint64
+	ObservedUnixNano int64
+	ReceiptUnixNano  int64
+	EventType        uint16
+	Stream           uint16
+	Compression      uint16
+	RawLength        uint64
+	Payload          []byte
+	ImmutableSHA256  [32]byte
+}
+
+type EventAppendResult struct {
+	ThroughEventSeq uint64
+	Duplicate       bool
+}
+
+type activeSegment struct {
+	segment   *Segment
+	ordinal   uint32
+	committed uint64
+}
+
+func (store *Store) AppendCommandEvent(ctx context.Context, event EventAppend) (EventAppendResult, error) {
+	var result EventAppendResult
+	if isZeroUUID(event.IssueUUID) || event.EventSeq == 0 || event.EventSeq > math.MaxInt64 || event.EventType == 0 || event.ReceiptUnixNano <= 0 || allZero(event.ImmutableSHA256[:]) {
+		return result, ErrInvalidSegmentRecord
+	}
+	record := SegmentRecord{
+		OwnerUUID: event.IssueUUID, Sequence: event.EventSeq, ObservedUnixNano: event.ObservedUnixNano,
+		ReceiptUnixNano: event.ReceiptUnixNano, Kind: PayloadKindCommandEvent, Stream: event.Stream,
+		Compression: event.Compression, RawLength: event.RawLength, Payload: event.Payload,
+	}
+	encoded, err := EncodeSegmentRecord(record, DefaultSegmentLimits())
+	if err != nil {
+		return result, err
+	}
+
+	store.writeMu.Lock()
+	defer store.writeMu.Unlock()
+	database, err := store.openDatabase()
+	if err != nil {
+		return result, err
+	}
+	var lastSequence uint64
+	err = database.QueryRowContext(ctx, `SELECT last_event_seq FROM commands WHERE issue_uuid = ?`, event.IssueUUID[:]).Scan(&lastSequence)
+	if errors.Is(err, sql.ErrNoRows) {
+		return result, ErrCommandNotFound
+	}
+	if err != nil {
+		return result, err
+	}
+	if event.EventSeq <= lastSequence {
+		var storedHash []byte
+		if err := database.QueryRowContext(ctx, `SELECT immutable_sha256 FROM command_events WHERE issue_uuid = ? AND event_seq = ?`, event.IssueUUID[:], event.EventSeq).Scan(&storedHash); err != nil {
+			return result, fmt.Errorf("load duplicate event: %w", err)
+		}
+		if len(storedHash) != len(event.ImmutableSHA256) || subtle.ConstantTimeCompare(storedHash, event.ImmutableSHA256[:]) != 1 {
+			return result, ErrEventConflict
+		}
+		return EventAppendResult{ThroughEventSeq: lastSequence, Duplicate: true}, nil
+	}
+	if event.EventSeq != lastSequence+1 {
+		return result, ErrEventSequenceGap
+	}
+
+	active, err := store.segmentForAppend(ctx, database, event.IssueUUID, uint64(len(encoded)))
+	if err != nil {
+		return result, err
+	}
+	recordOffset := active.segment.offset
+	committedEnd, err := active.segment.Append(record)
+	if err != nil {
+		store.invalidateActiveSegment(event.IssueUUID)
+		return result, err
+	}
+	if err := store.checkpoint(FaultAfterSegmentSyncBeforeMetadata); err != nil {
+		store.invalidateActiveSegment(event.IssueUUID)
+		return result, err
+	}
+
+	tx, err := database.BeginTx(ctx, nil)
+	if err != nil {
+		store.invalidateActiveSegment(event.IssueUUID)
+		return result, err
+	}
+	streamMix := uint64(0)
+	if event.Stream > 0 {
+		streamMix = uint64(1) << (event.Stream - 1)
+	}
+	if active.committed == 0 {
+		_, err = tx.ExecContext(ctx, `INSERT INTO output_segments (
+issue_uuid, ordinal, path, committed_end_offset, min_event_seq, max_event_seq, stream_mix,
+compressed_bytes, raw_bytes, checksum, sealed, created_at
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, 0, ?)`, event.IssueUUID[:], active.ordinal, active.segment.Name(), committedEnd,
+			event.EventSeq, event.EventSeq, streamMix, len(event.Payload), event.RawLength, event.ReceiptUnixNano)
+	} else {
+		var update sql.Result
+		update, err = tx.ExecContext(ctx, `UPDATE output_segments SET
+committed_end_offset = ?, max_event_seq = ?, stream_mix = stream_mix | ?,
+compressed_bytes = compressed_bytes + ?, raw_bytes = raw_bytes + ?
+WHERE issue_uuid = ? AND ordinal = ? AND committed_end_offset = ?`, committedEnd, event.EventSeq, streamMix,
+			len(event.Payload), event.RawLength, event.IssueUUID[:], active.ordinal, active.committed)
+		if err == nil {
+			var affected int64
+			affected, err = update.RowsAffected()
+			if err == nil && affected != 1 {
+				err = errors.New("segment committed offset changed unexpectedly")
+			}
+		}
+	}
+	if err == nil {
+		_, err = tx.ExecContext(ctx, `INSERT INTO command_events (
+issue_uuid, event_seq, observed_at, server_receipt_time, event_type, compression, raw_bytes, stored_bytes,
+payload, segment_ordinal, segment_record_offset, segment_record_length, immutable_sha256
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?)`, event.IssueUUID[:], event.EventSeq, event.ObservedUnixNano,
+			event.ReceiptUnixNano, event.EventType, event.Compression, event.RawLength, len(event.Payload), active.ordinal,
+			recordOffset, uint64(len(encoded)), event.ImmutableSHA256[:])
+	}
+	if err == nil {
+		var update sql.Result
+		update, err = tx.ExecContext(ctx, `UPDATE commands SET last_event_seq = ?, retained_compressed_bytes = retained_compressed_bytes + ?
+WHERE issue_uuid = ? AND last_event_seq = ?`, event.EventSeq, len(event.Payload), event.IssueUUID[:], lastSequence)
+		if err == nil {
+			var affected int64
+			affected, err = update.RowsAffected()
+			if err == nil && affected != 1 {
+				err = errors.New("command event sequence changed unexpectedly")
+			}
+		}
+	}
+	if err != nil {
+		_ = tx.Rollback()
+		store.invalidateActiveSegment(event.IssueUUID)
+		return result, fmt.Errorf("commit event metadata: %w", err)
+	}
+	if err := tx.Commit(); err != nil {
+		store.invalidateActiveSegment(event.IssueUUID)
+		return result, fmt.Errorf("commit event metadata: %w", err)
+	}
+	active.committed = committedEnd
+	result.ThroughEventSeq = event.EventSeq
+	if err := store.checkpoint(FaultAfterMetadataCommit); err != nil {
+		return result, err
+	}
+	return result, nil
+}
+
+func (store *Store) segmentForAppend(ctx context.Context, database *sql.DB, owner [16]byte, recordLength uint64) (*activeSegment, error) {
+	if current := store.activeSegments[owner]; current != nil {
+		if current.segment.offset == 0 || current.segment.offset+recordLength <= store.segmentTarget {
+			return current, nil
+		}
+		if err := current.segment.Close(); err != nil {
+			return nil, err
+		}
+		delete(store.activeSegments, owner)
+	}
+	var highest int64
+	if err := database.QueryRowContext(ctx, `SELECT COALESCE(MAX(ordinal), -1) FROM output_segments WHERE issue_uuid = ?`, owner[:]).Scan(&highest); err != nil {
+		return nil, err
+	}
+	if highest >= math.MaxUint32 {
+		return nil, errors.New("segment ordinal exhausted")
+	}
+	ordinal := uint32(highest + 1)
+	for {
+		segment, err := CreateSegment(store.segmentDirectory(), owner, ordinal, DefaultSegmentLimits())
+		if err == nil {
+			active := &activeSegment{segment: segment, ordinal: ordinal}
+			store.activeSegments[owner] = active
+			return active, nil
+		}
+		if !errors.Is(err, fs.ErrExist) || ordinal == math.MaxUint32 {
+			return nil, err
+		}
+		ordinal++
+	}
+}
+
+func (store *Store) invalidateActiveSegment(owner [16]byte) {
+	if active := store.activeSegments[owner]; active != nil {
+		_ = active.segment.Close()
+		delete(store.activeSegments, owner)
+	}
+}
+
+func (store *Store) openDatabase() (*sql.DB, error) {
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	if store.db == nil {
+		return nil, ErrStoreClosed
+	}
+	return store.db, nil
+}
+
+func (store *Store) checkpoint(name string) error {
+	if store.faultInjector == nil {
+		return nil
+	}
+	return store.faultInjector.Checkpoint(name)
+}
+
+func (store *Store) segmentDirectory() string { return filepath.Join(store.dataDir, "segments") }
@@ -102,8 +102,12 @@ CREATE TABLE command_events (
  event_seq INTEGER NOT NULL CHECK(event_seq > 0), observed_at INTEGER NOT NULL, server_receipt_time INTEGER NOT NULL,
  event_type INTEGER NOT NULL, compression INTEGER NOT NULL CHECK(compression IN (1,2)),
  raw_bytes INTEGER NOT NULL CHECK(raw_bytes >= 0), stored_bytes INTEGER NOT NULL CHECK(stored_bytes >= 0),
-  payload BLOB NOT NULL, immutable_sha256 BLOB NOT NULL CHECK(length(immutable_sha256) = 32),
-  PRIMARY KEY(issue_uuid, event_seq)
+  payload BLOB, segment_ordinal INTEGER, segment_record_offset INTEGER, segment_record_length INTEGER,
+  immutable_sha256 BLOB NOT NULL CHECK(length(immutable_sha256) = 32),
+  PRIMARY KEY(issue_uuid, event_seq),
+  CHECK((payload IS NOT NULL AND segment_ordinal IS NULL AND segment_record_offset IS NULL AND segment_record_length IS NULL) OR
+        (payload IS NULL AND segment_ordinal >= 0 AND segment_record_offset >= 0 AND segment_record_length > 0)),
+  FOREIGN KEY(issue_uuid, segment_ordinal) REFERENCES output_segments(issue_uuid, ordinal) ON DELETE CASCADE
 ) STRICT, WITHOUT ROWID;
 CREATE TABLE output_segments (
  issue_uuid BLOB NOT NULL REFERENCES commands(issue_uuid) ON DELETE CASCADE,
@@ -111,7 +115,8 @@ CREATE TABLE output_segments (
  min_event_seq INTEGER NOT NULL, max_event_seq INTEGER NOT NULL CHECK(max_event_seq >= min_event_seq),
  stream_mix INTEGER NOT NULL CHECK(stream_mix >= 0),
  compressed_bytes INTEGER NOT NULL CHECK(compressed_bytes >= 0), raw_bytes INTEGER NOT NULL CHECK(raw_bytes >= 0),
-  checksum BLOB NOT NULL CHECK(length(checksum) = 32), created_at INTEGER NOT NULL,
+  checksum BLOB CHECK(checksum IS NULL OR length(checksum) = 32), sealed INTEGER NOT NULL DEFAULT 0 CHECK(sealed IN (0,1)),
+  created_at INTEGER NOT NULL,
  PRIMARY KEY(issue_uuid, ordinal), UNIQUE(path)
 ) STRICT, WITHOUT ROWID;
 CREATE TABLE output_truncations (
@@ -0,0 +1,187 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+)
+
+var ErrIntegrityCheck = errors.New("SQLite integrity check failed")
+
+type RecoveryReport struct {
+	SegmentsChecked    uint64
+	RecordsChecked     uint64
+	TailBytesTruncated uint64
+	OrphansRemoved     uint64
+}
+
+func (store *Store) QuickCheck(ctx context.Context) error {
+	store.writeMu.Lock()
+	defer store.writeMu.Unlock()
+	database, err := store.openDatabase()
+	if err != nil {
+		return err
+	}
+	return store.quickCheckDatabase(ctx, database)
+}
+
+func (store *Store) RecoverCommandSegments(ctx context.Context) (RecoveryReport, error) {
+	var report RecoveryReport
+	store.writeMu.Lock()
+	defer store.writeMu.Unlock()
+	database, err := store.openDatabase()
+	if err != nil {
+		return report, err
+	}
+	if err := store.quickCheckDatabase(ctx, database); err != nil {
+		return report, err
+	}
+	rows, err := database.QueryContext(ctx, `SELECT issue_uuid, ordinal, path, committed_end_offset,
+min_event_seq, max_event_seq, stream_mix, compressed_bytes, raw_bytes
+FROM output_segments ORDER BY issue_uuid, ordinal`)
+	if err != nil {
+		return report, err
+	}
+	type segmentMetadata struct {
+		owner                                                                 [16]byte
+		ordinal                                                               uint32
+		name                                                                  string
+		committed, minSequence, maxSequence, streamMix, storedBytes, rawBytes uint64
+	}
+	var metadata []segmentMetadata
+	for rows.Next() {
+		var current segmentMetadata
+		var ownerBytes []byte
+		if err := rows.Scan(&ownerBytes, &current.ordinal, &current.name, &current.committed, &current.minSequence, &current.maxSequence, &current.streamMix, &current.storedBytes, &current.rawBytes); err != nil {
+			_ = rows.Close()
+			return report, err
+		}
+		if len(ownerBytes) != 16 {
+			_ = rows.Close()
+			return report, fmt.Errorf("%w: invalid segment owner", ErrCommittedRangeMissing)
+		}
+		copy(current.owner[:], ownerBytes)
+		metadata = append(metadata, current)
+	}
+	if err := rows.Close(); err != nil {
+		return report, err
+	}
+	referenced := map[string]bool{}
+	for _, current := range metadata {
+		nameOwner, nameOrdinal, err := parseSegmentName(current.name)
+		if err != nil || nameOwner != current.owner || nameOrdinal != current.ordinal {
+			return report, ErrUnsafeSegmentReference
+		}
+		recovery, err := RecoverSegment(store.segmentDirectory(), current.name, current.committed, DefaultSegmentLimits())
+		if err != nil {
+			return report, err
+		}
+		if err := validateRecoveredMetadata(ctx, database, current.owner, current.ordinal, recovery, current.minSequence, current.maxSequence, current.streamMix, current.storedBytes, current.rawBytes); err != nil {
+			return report, err
+		}
+		referenced[current.name] = true
+		report.SegmentsChecked++
+		report.RecordsChecked += uint64(len(recovery.Records))
+		report.TailBytesTruncated += recovery.TruncatedTail
+	}
+	entries, err := os.ReadDir(store.segmentDirectory())
+	if err != nil {
+		return report, err
+	}
+	directoryChanged := false
+	for _, entry := range entries {
+		name := entry.Name()
+		if referenced[name] {
+			continue
+		}
+		if _, _, err := parseSegmentName(name); err != nil {
+			return report, ErrUnsafeSegmentReference
+		}
+		path := filepath.Join(store.segmentDirectory(), name)
+		info, err := os.Lstat(path)
+		if err != nil {
+			return report, err
+		}
+		if !info.Mode().IsRegular() || info.Mode().Perm()&0o077 != 0 {
+			return report, ErrUnsafeSegmentReference
+		}
+		if err := os.Remove(path); err != nil {
+			return report, err
+		}
+		directoryChanged = true
+		report.OrphansRemoved++
+	}
+	if directoryChanged {
+		if err := syncDirectory(store.segmentDirectory()); err != nil {
+			return report, err
+		}
+	}
+	if _, err := database.ExecContext(ctx, `UPDATE output_segments SET sealed = 1 WHERE sealed = 0`); err != nil {
+		return report, err
+	}
+	return report, nil
+}
+
+func (store *Store) quickCheckDatabase(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: result unreadable", ErrIntegrityCheck)
+		}
+		count++
+		if result != "ok" {
+			return fmt.Errorf("%w: database reported an inconsistency", ErrIntegrityCheck)
+		}
+	}
+	if err := rows.Err(); err != nil || count != 1 {
+		return fmt.Errorf("%w: incomplete result", ErrIntegrityCheck)
+	}
+	return nil
+}
+
+func validateRecoveredMetadata(ctx context.Context, database *sql.DB, owner [16]byte, ordinal uint32, recovery SegmentRecovery, minSequence, maxSequence, streamMix, storedBytes, rawBytes uint64) error {
+	rows, err := database.QueryContext(ctx, `SELECT event_seq, segment_record_offset, segment_record_length,
+compression, raw_bytes, stored_bytes FROM command_events
+WHERE issue_uuid = ? AND segment_ordinal = ? ORDER BY segment_record_offset`, owner[:], ordinal)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	var index int
+	var computedStored, computedRaw, computedStreams uint64
+	for rows.Next() {
+		if index >= len(recovery.Records) {
+			return ErrCommittedRangeMissing
+		}
+		var sequence, offset, length, compression, raw, stored uint64
+		if err := rows.Scan(&sequence, &offset, &length, &compression, &raw, &stored); err != nil {
+			return err
+		}
+		recovered := recovery.Records[index]
+		if recovered.Offset != offset || recovered.Length != length || recovered.Record.OwnerUUID != owner || recovered.Record.Sequence != sequence || uint64(recovered.Record.Compression) != compression || recovered.Record.RawLength != raw || uint64(len(recovered.Record.Payload)) != stored {
+			return ErrCommittedRangeMissing
+		}
+		if recovered.Record.Stream > 0 {
+			computedStreams |= uint64(1) << (recovered.Record.Stream - 1)
+		}
+		computedStored += stored
+		computedRaw += raw
+		index++
+	}
+	if err := rows.Err(); err != nil {
+		return err
+	}
+	if index != len(recovery.Records) || index == 0 || recovery.Records[0].Record.Sequence != minSequence || recovery.Records[index-1].Record.Sequence != maxSequence || computedStreams != streamMix || computedStored != storedBytes || computedRaw != rawBytes {
+		return ErrCommittedRangeMissing
+	}
+	return nil
+}
@@ -292,12 +292,18 @@ func (segment *Segment) Close() error {
 }

 type SegmentRecovery struct {
-	Records         []SegmentRecord
+	Records         []RecoveredSegmentRecord
 	CommittedOffset uint64
 	OriginalSize    uint64
 	TruncatedTail   uint64
 }

+type RecoveredSegmentRecord struct {
+	Offset uint64
+	Length uint64
+	Record SegmentRecord
+}
+
 func RecoverSegment(directory, name string, committedOffset uint64, limits SegmentLimits) (SegmentRecovery, error) {
 	var result SegmentRecovery
 	if filepath.Base(name) != name {
@@ -338,7 +344,7 @@ func RecoverSegment(directory, name string, committedOffset uint64, limits Segme
 			return result, ErrCommittedRangeMissing
 		}
 		consumed += length
-		result.Records = append(result.Records, record)
+		result.Records = append(result.Records, RecoveredSegmentRecord{Offset: consumed - length, Length: length, Record: record})
 	}
 	if consumed != committedOffset {
 		return result, ErrCommittedRangeMissing
@@ -22,14 +22,21 @@ var (
 )

 type Options struct {
-	DataDir     string
-	BusyTimeout time.Duration
+	DataDir           string
+	BusyTimeout       time.Duration
+	SegmentTargetSize uint64
+	FaultInjector     FaultInjector
 }

 type Store struct {
-	db     *sql.DB
-	unlock func() error
-	mu     sync.Mutex
+	db             *sql.DB
+	unlock         func() error
+	dataDir        string
+	segmentTarget  uint64
+	faultInjector  FaultInjector
+	activeSegments map[[16]byte]*activeSegment
+	mu             sync.Mutex
+	writeMu        sync.Mutex
 }

 func Open(ctx context.Context, options Options) (*Store, error) {
@@ -39,6 +46,12 @@ func Open(ctx context.Context, options Options) (*Store, error) {
 	if options.BusyTimeout <= 0 {
 		return nil, fmt.Errorf("busy timeout must be positive")
 	}
+	if options.SegmentTargetSize == 0 {
+		options.SegmentTargetSize = 256 << 10
+	}
+	if options.SegmentTargetSize > DefaultSegmentLimit {
+		return nil, fmt.Errorf("segment target exceeds hard record limit")
+	}
 	if err := ensurePrivateDirectory(options.DataDir); err != nil {
 		return nil, err
 	}
@@ -73,7 +86,10 @@ func Open(ctx context.Context, options Options) (*Store, error) {
 	}
 	db.SetMaxOpenConns(1)
 	db.SetMaxIdleConns(1)
-	store := &Store{db: db, unlock: unlock}
+	store := &Store{
+		db: db, unlock: unlock, dataDir: options.DataDir, segmentTarget: options.SegmentTargetSize,
+		faultInjector: options.FaultInjector, activeSegments: make(map[[16]byte]*activeSegment),
+	}
 	if err := db.PingContext(ctx); err != nil {
 		_ = store.Close()
 		return nil, fmt.Errorf("open SQLite: %w", err)
@@ -85,15 +101,27 @@ func Open(ctx context.Context, options Options) (*Store, error) {
 	return store, nil
 }

-func (store *Store) DB() *sql.DB { return store.db }
+func (store *Store) DB() *sql.DB {
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	return store.db
+}

 func (store *Store) Close() error {
 	if store == nil {
 		return nil
 	}
+	store.writeMu.Lock()
+	defer store.writeMu.Unlock()
 	store.mu.Lock()
 	defer store.mu.Unlock()
 	var result error
+	for owner, active := range store.activeSegments {
+		if err := active.segment.Close(); result == nil && err != nil {
+			result = err
+		}
+		delete(store.activeSegments, owner)
+	}
 	if store.db != nil {
 		result = store.db.Close()
 		store.db = nil