feat: add crash-safe segment framing

test/integration/store/store_integration_test.go

@@ -207,6 +207,177 @@ func TestMigrationChecksumMismatchPreventsOpen_REC_STORE_01(t *testing.T) {
 	}
 }
 
+func TestSegmentAppendTailRecoveryAndCommittedCorruption_REC_STORE_02(t *testing.T) {
+	t.Parallel()
+
+	directory := filepath.Join(t.TempDir(), "segments")
+	if err := os.Mkdir(directory, 0o700); err != nil {
+		t.Fatal(err)
+	}
+	owner := uuidBytes(10)
+	segment, err := store.CreateSegment(directory, owner, 1, store.DefaultSegmentLimits())
+	if err != nil {
+		t.Fatal(err)
+	}
+	name := segment.Name()
+	first := integrationRecord(owner, 1, "first")
+	second := integrationRecord(owner, 2, "second")
+	if _, err := segment.Append(first); err != nil {
+		t.Fatal(err)
+	}
+	committed, err := segment.Append(second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := segment.Close(); err != nil {
+		t.Fatal(err)
+	}
+	path := filepath.Join(directory, name)
+	file, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := file.Write([]byte("unacknowledged-tail")); err != nil {
+		t.Fatal(err)
+	}
+	if err := file.Sync(); err != nil {
+		t.Fatal(err)
+	}
+	if err := file.Close(); err != nil {
+		t.Fatal(err)
+	}
+	recovery, err := store.RecoverSegment(directory, name, committed, store.DefaultSegmentLimits())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if recovery.TruncatedTail != uint64(len("unacknowledged-tail")) || len(recovery.Records) != 2 {
+		t.Fatalf("recovery = %+v", recovery)
+	}
+	info, err := os.Stat(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if uint64(info.Size()) != committed {
+		t.Fatalf("size = %d, want %d", info.Size(), committed)
+	}
+
+	corrupt, err := os.OpenFile(path, os.O_RDWR, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := corrupt.WriteAt([]byte{0xff}, 128); err != nil {
+		t.Fatal(err)
+	}
+	if err := corrupt.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := store.RecoverSegment(directory, name, committed, store.DefaultSegmentLimits()); !errors.Is(err, store.ErrSegmentChecksum) {
+		t.Fatalf("corrupt committed range error = %v", err)
+	}
+	info, err = os.Stat(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if uint64(info.Size()) != committed {
+		t.Fatalf("corruption changed committed file: size=%d", info.Size())
+	}
+}
+
+func TestSegmentShortRangeUnsafeReferenceAndExclusiveCreate_BH_STORE_05(t *testing.T) {
+	t.Parallel()
+
+	directory := filepath.Join(t.TempDir(), "segments")
+	if err := os.Mkdir(directory, 0o700); err != nil {
+		t.Fatal(err)
+	}
+	owner := uuidBytes(20)
+	segment, err := store.CreateSegment(directory, owner, 9, store.DefaultSegmentLimits())
+	if err != nil {
+		t.Fatal(err)
+	}
+	name := segment.Name()
+	committed, err := segment.Append(integrationRecord(owner, 1, "payload"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := store.CreateSegment(directory, owner, 9, store.DefaultSegmentLimits()); err == nil {
+		t.Fatal("exclusive segment name collision succeeded")
+	}
+	if err := segment.Close(); err != nil {
+		t.Fatal(err)
+	}
+	path := filepath.Join(directory, name)
+	if err := os.Truncate(path, int64(committed-1)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := store.RecoverSegment(directory, name, committed, store.DefaultSegmentLimits()); !errors.Is(err, store.ErrCommittedRangeMissing) {
+		t.Fatalf("short committed range error = %v", err)
+	}
+	for _, unsafe := range []string{"../" + name, "not-a-segment", filepath.Join("nested", name)} {
+		if _, err := store.RecoverSegment(directory, unsafe, 0, store.DefaultSegmentLimits()); !errors.Is(err, store.ErrUnsafeSegmentReference) {
+			t.Errorf("unsafe reference %q error = %v", unsafe, err)
+		}
+	}
+}
+
+func TestSegmentConcurrentAppendSerializesRecords_RACE_STORE_02(t *testing.T) {
+	t.Parallel()
+
+	directory := filepath.Join(t.TempDir(), "segments")
+	if err := os.Mkdir(directory, 0o700); err != nil {
+		t.Fatal(err)
+	}
+	owner := uuidBytes(30)
+	segment, err := store.CreateSegment(directory, owner, 1, store.DefaultSegmentLimits())
+	if err != nil {
+		t.Fatal(err)
+	}
+	const count = 16
+	ends := make(chan uint64, count)
+	errorsFound := make(chan error, count)
+	var wait sync.WaitGroup
+	for sequence := 1; sequence <= count; sequence++ {
+		wait.Add(1)
+		go func(sequence uint64) {
+			defer wait.Done()
+			end, err := segment.Append(integrationRecord(owner, sequence, "parallel"))
+			if err != nil {
+				errorsFound <- err
+				return
+			}
+			ends <- end
+		}(uint64(sequence))
+	}
+	wait.Wait()
+	close(errorsFound)
+	for err := range errorsFound {
+		t.Fatal(err)
+	}
+	close(ends)
+	var committed uint64
+	seenEnds := map[uint64]bool{}
+	for end := range ends {
+		if seenEnds[end] {
+			t.Fatalf("duplicate committed end %d", end)
+		}
+		seenEnds[end] = true
+		if end > committed {
+			committed = end
+		}
+	}
+	name := segment.Name()
+	if err := segment.Close(); err != nil {
+		t.Fatal(err)
+	}
+	recovery, err := store.RecoverSegment(directory, name, committed, store.DefaultSegmentLimits())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(recovery.Records) != count {
+		t.Fatalf("record count = %d, want %d", len(recovery.Records), count)
+	}
+}
+
 func openStore(t *testing.T, dataDir string) *store.Store {
 	t.Helper()
 	opened, err := store.Open(context.Background(), store.Options{DataDir: dataDir, BusyTimeout: busyTimeout})
@@ -234,3 +405,18 @@ func bytesOf(length int, value byte) []byte {
 	}
 	return result
 }
+
+func uuidBytes(start byte) [16]byte {
+	var result [16]byte
+	for index := range result {
+		result[index] = start + byte(index)
+	}
+	return result
+}
+
+func integrationRecord(owner [16]byte, sequence uint64, payload string) store.SegmentRecord {
+	return store.SegmentRecord{
+		OwnerUUID: owner, Sequence: sequence, ObservedUnixNano: int64(sequence), ReceiptUnixNano: int64(sequence + 1),
+		Kind: store.PayloadKindCommandEvent, Stream: 1, Compression: 1, RawLength: uint64(len(payload)), Payload: []byte(payload),
+	}
+}