diff --git a/internal/server/store/event.go b/internal/server/store/event.go new file mode 100644 index 0000000..165f7dc --- /dev/null +++ b/internal/server/store/event.go @@ -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") } diff --git a/internal/server/store/migrations.go b/internal/server/store/migrations.go index dd11371..5816c0a 100644 --- a/internal/server/store/migrations.go +++ b/internal/server/store/migrations.go @@ -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 ( diff --git a/internal/server/store/recovery.go b/internal/server/store/recovery.go new file mode 100644 index 0000000..61474ce --- /dev/null +++ b/internal/server/store/recovery.go @@ -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, ¤t.ordinal, ¤t.name, ¤t.committed, ¤t.minSequence, ¤t.maxSequence, ¤t.streamMix, ¤t.storedBytes, ¤t.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 +} diff --git a/internal/server/store/segment.go b/internal/server/store/segment.go index be4fa4a..b3d1531 100644 --- a/internal/server/store/segment.go +++ b/internal/server/store/segment.go @@ -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 diff --git a/internal/server/store/store.go b/internal/server/store/store.go index 2099083..3b4727e 100644 --- a/internal/server/store/store.go +++ b/internal/server/store/store.go @@ -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 diff --git a/test/coverage.toml b/test/coverage.toml index b8e14af..bd88b8b 100644 --- a/test/coverage.toml +++ b/test/coverage.toml @@ -237,3 +237,27 @@ id = "RACE-STORE-02" layer = "integration" status = "implemented" tests = ["test/integration/store/store_integration_test.go:TestSegmentConcurrentAppendSerializesRecords_RACE_STORE_02"] + +[[requirements]] +id = "CRASH-STORE-01" +layer = "integration" +status = "implemented" +tests = ["test/integration/store/store_integration_test.go:TestEventCrashBeforeMetadataDropsUncommittedTail_CRASH_STORE_01"] + +[[requirements]] +id = "CRASH-STORE-02" +layer = "integration" +status = "implemented" +tests = ["test/integration/store/store_integration_test.go:TestEventCrashAfterCommitReplaysWithoutDuplication_CRASH_STORE_02"] + +[[requirements]] +id = "HP-STORE-03" +layer = "integration" +status = "implemented" +tests = ["test/integration/store/store_integration_test.go:TestEventGapRejectedBeforeFileWriteAndRotation_HP_STORE_03"] + +[[requirements]] +id = "CRASH-STORE-03" +layer = "integration" +status = "implemented" +tests = ["test/integration/store/store_integration_test.go:TestEventCrashOnExistingSegmentTruncatesOnlyTail_CRASH_STORE_03"] diff --git a/test/integration/store/store_integration_test.go b/test/integration/store/store_integration_test.go index 28d12cc..405df66 100644 --- a/test/integration/store/store_integration_test.go +++ b/test/integration/store/store_integration_test.go @@ -4,6 +4,7 @@ package store_test import ( "context" + "crypto/sha256" "database/sql" "errors" "os" @@ -378,11 +379,168 @@ func TestSegmentConcurrentAppendSerializesRecords_RACE_STORE_02(t *testing.T) { } } +func TestEventCrashBeforeMetadataDropsUncommittedTail_CRASH_STORE_01(t *testing.T) { + t.Parallel() + + dataDir := filepath.Join(t.TempDir(), "state") + injected := errors.New("injected crash before metadata") + opened := openStoreWithOptions(t, store.Options{ + DataDir: dataDir, BusyTimeout: busyTimeout, + FaultInjector: &failOnceInjector{target: store.FaultAfterSegmentSyncBeforeMetadata, failure: injected}, + }) + issue := uuidBytes(40) + seedCommand(t, opened.DB(), issue) + event := appendEvent(issue, 1, "not-acknowledged") + if result, err := opened.AppendCommandEvent(context.Background(), event); !errors.Is(err, injected) || result.ThroughEventSeq != 0 { + t.Fatalf("AppendCommandEvent = (%+v, %v)", result, err) + } + assertCommandEventState(t, opened.DB(), issue, 0, 0) + if err := opened.Close(); err != nil { + t.Fatal(err) + } + + reopened := openStore(t, dataDir) + recovery, err := reopened.RecoverCommandSegments(context.Background()) + if err != nil { + t.Fatal(err) + } + if recovery.OrphansRemoved != 1 || recovery.RecordsChecked != 0 { + t.Fatalf("recovery = %+v", recovery) + } + result, err := reopened.AppendCommandEvent(context.Background(), event) + if err != nil || result.ThroughEventSeq != 1 || result.Duplicate { + t.Fatalf("retry = (%+v, %v)", result, err) + } + assertCommandEventState(t, reopened.DB(), issue, 1, 1) + if err := reopened.Close(); err != nil { + t.Fatal(err) + } +} + +func TestEventCrashAfterCommitReplaysWithoutDuplication_CRASH_STORE_02(t *testing.T) { + t.Parallel() + + dataDir := filepath.Join(t.TempDir(), "state") + injected := errors.New("injected lost acknowledgement") + opened := openStoreWithOptions(t, store.Options{ + DataDir: dataDir, BusyTimeout: busyTimeout, + FaultInjector: &failOnceInjector{target: store.FaultAfterMetadataCommit, failure: injected}, + }) + issue := uuidBytes(60) + seedCommand(t, opened.DB(), issue) + event := appendEvent(issue, 1, "committed") + if result, err := opened.AppendCommandEvent(context.Background(), event); !errors.Is(err, injected) || result.ThroughEventSeq != 1 { + t.Fatalf("AppendCommandEvent = (%+v, %v)", result, err) + } + assertCommandEventState(t, opened.DB(), issue, 1, 1) + if err := opened.Close(); err != nil { + t.Fatal(err) + } + + reopened := openStore(t, dataDir) + recovery, err := reopened.RecoverCommandSegments(context.Background()) + if err != nil { + t.Fatal(err) + } + if recovery.SegmentsChecked != 1 || recovery.RecordsChecked != 1 || recovery.OrphansRemoved != 0 { + t.Fatalf("recovery = %+v", recovery) + } + result, err := reopened.AppendCommandEvent(context.Background(), event) + if err != nil || !result.Duplicate || result.ThroughEventSeq != 1 { + t.Fatalf("duplicate retry = (%+v, %v)", result, err) + } + conflict := event + conflict.ImmutableSHA256 = sha256.Sum256([]byte("different immutable event")) + if _, err := reopened.AppendCommandEvent(context.Background(), conflict); !errors.Is(err, store.ErrEventConflict) { + t.Fatalf("conflicting duplicate error = %v", err) + } + assertCommandEventState(t, reopened.DB(), issue, 1, 1) + if err := reopened.Close(); err != nil { + t.Fatal(err) + } +} + +func TestEventCrashOnExistingSegmentTruncatesOnlyTail_CRASH_STORE_03(t *testing.T) { + t.Parallel() + + dataDir := filepath.Join(t.TempDir(), "state") + injected := errors.New("injected second append crash") + opened := openStoreWithOptions(t, store.Options{ + DataDir: dataDir, BusyTimeout: busyTimeout, + FaultInjector: &failOnceInjector{target: store.FaultAfterSegmentSyncBeforeMetadata, failure: injected, skip: 1}, + }) + issue := uuidBytes(70) + seedCommand(t, opened.DB(), issue) + if _, err := opened.AppendCommandEvent(context.Background(), appendEvent(issue, 1, "committed-first")); err != nil { + t.Fatal(err) + } + second := appendEvent(issue, 2, "uncommitted-second") + if _, err := opened.AppendCommandEvent(context.Background(), second); !errors.Is(err, injected) { + t.Fatalf("second append error = %v", err) + } + assertCommandEventState(t, opened.DB(), issue, 1, 1) + if err := opened.Close(); err != nil { + t.Fatal(err) + } + + reopened := openStore(t, dataDir) + recovery, err := reopened.RecoverCommandSegments(context.Background()) + if err != nil { + t.Fatal(err) + } + if recovery.SegmentsChecked != 1 || recovery.RecordsChecked != 1 || recovery.TailBytesTruncated == 0 || recovery.OrphansRemoved != 0 { + t.Fatalf("recovery = %+v", recovery) + } + if result, err := reopened.AppendCommandEvent(context.Background(), second); err != nil || result.ThroughEventSeq != 2 { + t.Fatalf("second retry = (%+v, %v)", result, err) + } + assertCommandEventState(t, reopened.DB(), issue, 2, 2) + if err := reopened.Close(); err != nil { + t.Fatal(err) + } +} + +func TestEventGapRejectedBeforeFileWriteAndRotation_HP_STORE_03(t *testing.T) { + t.Parallel() + + dataDir := filepath.Join(t.TempDir(), "state") + opened := openStoreWithOptions(t, store.Options{DataDir: dataDir, BusyTimeout: busyTimeout, SegmentTargetSize: 200}) + issue := uuidBytes(80) + seedCommand(t, opened.DB(), issue) + if _, err := opened.AppendCommandEvent(context.Background(), appendEvent(issue, 2, "gap")); !errors.Is(err, store.ErrEventSequenceGap) { + t.Fatalf("gap error = %v", err) + } + entries, err := os.ReadDir(filepath.Join(dataDir, "segments")) + if err != nil || len(entries) != 0 { + t.Fatalf("segments after gap = %v, err = %v", entries, err) + } + for sequence, payload := range []string{"one", "two"} { + if _, err := opened.AppendCommandEvent(context.Background(), appendEvent(issue, uint64(sequence+1), payload)); err != nil { + t.Fatal(err) + } + } + var segmentCount int + if err := opened.DB().QueryRow(`SELECT count(*) FROM output_segments WHERE issue_uuid = ?`, issue[:]).Scan(&segmentCount); err != nil || segmentCount != 2 { + t.Fatalf("segment count = %d, err = %v", segmentCount, err) + } + if err := opened.QuickCheck(context.Background()); err != nil { + t.Fatal(err) + } + if err := opened.Close(); err != nil { + t.Fatal(err) + } +} + func openStore(t *testing.T, dataDir string) *store.Store { t.Helper() - opened, err := store.Open(context.Background(), store.Options{DataDir: dataDir, BusyTimeout: busyTimeout}) + return openStoreWithOptions(t, store.Options{DataDir: dataDir, BusyTimeout: busyTimeout}) +} + +func openStoreWithOptions(t *testing.T, options store.Options) *store.Store { + t.Helper() + opened, err := store.Open(context.Background(), options) if err != nil { - t.Fatalf("Open(%s): %v", dataDir, err) + t.Fatalf("Open(%s): %v", options.DataDir, err) } return opened } @@ -420,3 +578,65 @@ func integrationRecord(owner [16]byte, sequence uint64, payload string) store.Se Kind: store.PayloadKindCommandEvent, Stream: 1, Compression: 1, RawLength: uint64(len(payload)), Payload: []byte(payload), } } + +type failOnceInjector struct { + mu sync.Mutex + target string + failure error + skip int + fired bool +} + +func (injector *failOnceInjector) Checkpoint(name string) error { + injector.mu.Lock() + defer injector.mu.Unlock() + if name == injector.target && !injector.fired { + if injector.skip > 0 { + injector.skip-- + return nil + } + injector.fired = true + return injector.failure + } + return nil +} + +func seedCommand(t *testing.T, database *sql.DB, issue [16]byte) { + t.Helper() + clientInstance := uuidBytes(100) + if _, err := database.Exec(`INSERT OR IGNORE INTO clients ( +client_id, platform, architecture, daemon_version, daemon_cwd, supported_shells, capabilities, client_instance_id +) VALUES ('client-a', 3, 'amd64', 'test', 'C:\\work', x'01', x'', ?)`, clientInstance[:]); err != nil { + t.Fatal(err) + } + requestHash := sha256.Sum256([]byte("request")) + if _, err := database.Exec(`INSERT INTO commands ( +issue_uuid, client_id, issue_time, server_receipt_time, lifecycle, revision, immutable_request_sha256, +execution_spec, execution_spec_raw_bytes, execution_spec_stored_bytes, execution_spec_compression +) VALUES (?, 'client-a', 1, 1, 1, 1, ?, x'01', 1, 1, 1)`, issue[:], requestHash[:]); err != nil { + t.Fatal(err) + } +} + +func appendEvent(issue [16]byte, sequence uint64, payload string) store.EventAppend { + immutableHash := sha256.Sum256(append([]byte{byte(sequence)}, []byte(payload)...)) + return store.EventAppend{ + IssueUUID: issue, EventSeq: sequence, ObservedUnixNano: int64(sequence), ReceiptUnixNano: int64(sequence + 10), + EventType: 1, Stream: 1, Compression: 1, RawLength: uint64(len(payload)), Payload: []byte(payload), + ImmutableSHA256: immutableHash, + } +} + +func assertCommandEventState(t *testing.T, database *sql.DB, issue [16]byte, wantLast, wantEvents int) { + t.Helper() + var last, events int + if err := database.QueryRow(`SELECT last_event_seq FROM commands WHERE issue_uuid = ?`, issue[:]).Scan(&last); err != nil { + t.Fatal(err) + } + if err := database.QueryRow(`SELECT count(*) FROM command_events WHERE issue_uuid = ?`, issue[:]).Scan(&events); err != nil { + t.Fatal(err) + } + if last != wantLast || events != wantEvents { + t.Fatalf("event state = (last=%d, rows=%d), want (%d, %d)", last, events, wantLast, wantEvents) + } +}