diff --git a/correctness_faults_test.go b/correctness_faults_test.go new file mode 100644 index 0000000..c79b53a --- /dev/null +++ b/correctness_faults_test.go @@ -0,0 +1,270 @@ +// SPDX-License-Identifier: MIT + +//go:build diskqueue_faults + +package diskqueue + +import ( + "errors" + "os" + "path/filepath" + "testing" +) + +func TestExplicitFlushPersistsNoSyncDirectory(t *testing.T) { + for _, closeQueue := range []bool{false, true} { + w, dir := openAckQueue(t, ackOpts()) + addN(t, w, 1) + injectAt(t, "sync.dir", errInjected) + flush := w.Sync + if closeQueue { + flush = w.Close + } + if err := flush(); !errors.Is(err, errInjected) || !errors.Is(err, ErrIO) { + t.Errorf("Close=%v: directory flush failure was not reported: %v", closeQueue, err) + } + if w.Stats().UnsyncedBytes == 0 { + t.Error("failed directory flush reported all bytes durable") + } + faultHook = nil + _ = w.Close() + reopened, err := New[uint64](dir, marshalU64, unmarshalU64, ackOpts()) + if err != nil { + t.Fatal(err) + } + if v, ok, err := reopened.NewReader().TryTake(); err != nil || !ok || v != 0 { + t.Fatalf("record after reopen: %d, %v, %v", v, ok, err) + } + if err := reopened.Close(); err != nil { + t.Fatal(err) + } + } +} + +func TestZeroScanReadErrorPreservesSegment(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "data.00000001") + if err := os.WriteFile(path, make([]byte, 4096+headerSize), 0o644); err != nil { + t.Fatal(err) + } + injectAt(t, "recovery.readZero", errInjected) + w, err := New[uint64](dir, marshalU64, unmarshalU64, ackOpts()) + if err == nil { + _ = w.Close() + t.Error("recovery succeeded after failing to read the zero-filled segment") + } else if !errors.Is(err, errInjected) { + t.Errorf("recovery error lost its cause: %v", err) + } + if _, err := os.Stat(path); err != nil { + t.Fatalf("failed read removed the segment: %v", err) + } +} + +func TestFailedTruncationRecoveryClosesHandles(t *testing.T) { + dir := t.TempDir() + opts := manyRecords(t, dir, 700) + path := filepath.Join(dir, "data.00000001") + if err := os.Truncate(path, headerSize+70003); err != nil { + t.Fatal(err) + } + failReadsAfter(t, 0, errInjected) + before := openFDs(t) + for i := 0; i < 5; i++ { + w, err := New[uint64](dir, bigM(), bigU(), opts) + if err == nil { + _ = w.Close() + t.Fatal("recovery unexpectedly succeeded") + } + } + if after := openFDs(t); after > before { + t.Fatalf("failed recovery leaked %d descriptors", after-before) + } +} + +func TestGroupFlushReportsInterveningDurabilityFailure(t *testing.T) { + for _, point := range []string{"append.syncData", "append.syncHeader"} { + w, _ := openAckQueue(t, Options{SegmentSize: 4096}) + faultHook = func(name string) error { + if name == point { + // Model a failure latched by a commit while the group flush has + // released the mutex. Its own fdatasync can still succeed. + w.st.failIO(errInjected) + } + return nil + } + err := w.Add(1) + faultHook = nil + if !errors.Is(err, errInjected) || !errors.Is(err, ErrIO) { + t.Errorf("%s: Add reported %v after an intervening durability failure", point, err) + } + _ = w.Close() + } +} + +func TestSyncKeepsDescriptorCapDuringConcurrentReads(t *testing.T) { + m, u := bytesCodec() + w, err := New[[]byte](t.TempDir(), m, u, Options{ + NoSync: true, SegmentSize: 4096, MaxSegments: -1, MaxOpenFiles: 3, + }) + if err != nil { + t.Fatal(err) + } + defer func() { _ = w.Close() }() + for i := 0; i < 8; i++ { + if err := w.Add(make([]byte, 3000)); err != nil { + t.Fatal(err) + } + } + entered, release := parkAtSyncFile(t) + done := make(chan error, 1) + go func() { done <- w.Sync() }() + <-entered + // The first chunk is pinned; read a segment outside that chunk while the + // active file also occupies a descriptor. + w.mu.Lock() + _, _, _, _, readErr := w.st.read(w.st.files[4].base) + open := w.st.nOpen + w.mu.Unlock() + close(release) + if err := <-done; err != nil { + t.Fatal(err) + } + if readErr != nil { + t.Fatal(readErr) + } + if open > 3 { + t.Fatalf("concurrent read opened %d descriptors under a cap of 3", open) + } +} + +func TestInterruptedRepairKeepsReplayDecision(t *testing.T) { + dir := emptyTornStore(t) + repairs := 0 + faultHook = func(name string) error { + if name == "recovery.writeHeader" { + repairs++ + if repairs == 2 { + return errInjected + } + } + return nil + } + s, err := openStore(dir, 32, 0, false, 0, 0) + faultHook = nil + if err == nil { + _ = s.close() + t.Fatal("open ignored a failed repair") + } + if !errors.Is(err, errInjected) { + t.Fatalf("open: %v", err) + } + s, err = openStore(dir, 32, 0, false, 0, 0) + if err != nil { + t.Fatal(err) + } + defer func() { _ = s.close() }() + if s.count() != 2 { + t.Fatalf("interrupted repair lost its replay decision: count=%d", s.count()) + } +} + +func TestRecoveryFlushesRepairsUnderNoSync(t *testing.T) { + dir := emptyTornStore(t) + injectAt(t, "recovery.syncHeader", errInjected) + s, err := openStore(dir, 32, 0, true, 0, 0) + faultHook = nil + if err == nil { + _ = s.close() + t.Fatal("NoSync let recovery publish repairs without flushing them") + } + if !errors.Is(err, ErrIO) || !errors.Is(err, errInjected) { + t.Fatalf("repair flush failure: %v", err) + } + s, err = openStore(dir, 32, 0, true, 0, 0) + if err != nil { + t.Fatal(err) + } + defer func() { _ = s.close() }() + if s.count() != 2 { + t.Fatalf("retry lost records after the failed repair flush: %d", s.count()) + } +} + +func TestFailedGroupPublicationWakesCapacityWaiters(t *testing.T) { + w, _ := openAckQueue(t, Options{SegmentSize: 4096, MaxBytes: 17}) + // While this span is in flight, its bytes exhaust the budget and AddWait + // can park on spaceNotify even though the visible queue is still empty. + if err := w.st.stagePending(make([]byte, 8)); err != nil { + t.Fatal(err) + } + waiter := make(chan struct{}) + w.spaceNotify = waiter + injectAt(t, "append.writeHeader", errInjected) + w.mu.Lock() + err := w.leadFlushLocked() + w.mu.Unlock() + if !errors.Is(err, errInjected) { + t.Fatalf("flush: %v", err) + } + select { + case <-waiter: + default: + t.Fatal("discarding unpublished bytes left capacity waiters asleep") + } + faultHook = nil + if err := w.Add(2); err != nil { + t.Fatalf("capacity was not released: %v", err) + } +} + +func TestSyncFailureWakesCapacityWaiters(t *testing.T) { + w, _ := openAckQueue(t, Options{NoSync: true, MaxBytes: 17}) + addN(t, w, 1) + waiter := make(chan struct{}) + w.spaceNotify = waiter + injectAt(t, "sync.file", errInjected) + if err := w.Sync(); !errors.Is(err, ErrIO) { + t.Fatalf("Sync: %v", err) + } + select { + case <-waiter: + default: + t.Fatal("a latched sync failure left capacity waiters asleep") + } + faultHook = nil +} + +func TestSyncSkipsFilesReclaimedBehindPinnedPrefix(t *testing.T) { + m, u := bytesCodec() + w, err := New[[]byte](t.TempDir(), m, u, Options{ + NoSync: true, SegmentSize: 4096, MaxSegments: -1, MaxOpenFiles: 3, + }) + if err != nil { + t.Fatal(err) + } + defer func() { _ = w.Close() }() + for i := 0; i < 8; i++ { + if err := w.Add(make([]byte, 3000)); err != nil { + t.Fatal(err) + } + } + entered, release := parkAtSyncFile(t) + done := make(chan error, 1) + go func() { done <- w.Sync() }() + <-entered + r := w.NewReader() + for i := 0; i < 7; i++ { + if _, ok, err := r.TryTake(); !ok || err != nil { + close(release) + _ = <-done + t.Fatalf("read %d: %v, %v", i, ok, err) + } + } + close(release) + if err := <-done; err != nil { + t.Fatalf("Sync tried to reopen a reclaimed file behind the pinned prefix: %v", err) + } + if w.Count() != 1 { + t.Fatalf("concurrent reclamation changed the unread record: %+v", w.Stats()) + } +} diff --git a/correctness_test.go b/correctness_test.go new file mode 100644 index 0000000..bdb797b --- /dev/null +++ b/correctness_test.go @@ -0,0 +1,485 @@ +// SPDX-License-Identifier: MIT + +package diskqueue + +import ( + "bytes" + "errors" + "os" + "testing" +) + +func TestCorruptionRetirementWaitsForReservations(t *testing.T) { + for _, framing := range []bool{false, true} { + name := "checksum" + if framing { + name = "framing" + } + t.Run(name, func(t *testing.T) { + w, dir := openAckQueue(t, ackOpts()) + addN(t, w, 3) + at, damage := 17+1, []byte{0xff} + if framing { + at, damage = 17, unframeable + } + corruptData(t, w.st, 0, at, damage) + r := w.NewReader() + off := reserveN(t, r, 1)[0] + if _, _, _, err := r.TryReserve(); !errors.Is(err, ErrCorrupt) { + t.Fatalf("damaged record: %v", err) + } + if w.st.commitOff != 0 { + t.Fatalf("corruption retired an outstanding reservation: commit=%d", w.st.commitOff) + } + if !framing { + tail := reserveN(t, r, 1)[0] + if err := r.Ack(tail); err != nil { + t.Fatal(err) + } + } + // A new append can share the damaged segment, including publication + // by a group flush already in flight when the damage was detected. + if err := w.Add(99); err != nil { + t.Fatal(err) + } + v, ok, later, err := r.TryReserve() + if err != nil || !ok || v != 99 { + t.Fatalf("new record behind damage: %d, %v, %v", v, ok, err) + } + if err := r.Ack(off); err != nil { + t.Fatal(err) + } + if w.Count() != 1 { + t.Fatalf("retiring damage also retired the later reservation: %+v", w.Stats()) + } + if err := r.Ack(later); err != nil { + t.Fatal(err) + } + if w.Count() != 0 || w.Size() != 0 || !w.Empty() { + t.Fatalf("ack left damaged backlog stranded: %+v", w.Stats()) + } + if got := w.Stats().Corruptions; got != 1 { + t.Fatalf("one damaged region reported %d times", got) + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + reopened, err := New[uint64](dir, marshalU64, unmarshalU64, ackOpts()) + if err != nil { + t.Fatal(err) + } + defer func() { _ = reopened.Close() }() + if reopened.Count() != 0 || !reopened.Empty() { + t.Fatalf("retired damage replayed: %+v", reopened.Stats()) + } + }) + } +} + +func TestAcknowledgementAfterSegmentDisappears(t *testing.T) { + w, dir := openAckQueue(t, ackOpts()) + addN(t, w, 245) + r := w.NewReader() + first := reserveN(t, r, 1)[0] + gone := w.st.files[0] + quarantineEvictHandle(t, w.st, 0) + if err := os.Remove(gone.path); err != nil { + t.Fatal(err) + } + if _, _, _, err := r.TryReserve(); !errors.Is(err, ErrCorrupt) { + t.Fatalf("missing segment: %v", err) + } + if w.st.commitOff != 0 { + t.Fatal("missing segment retired an outstanding reservation") + } + if err := r.Ack(first); err != nil { + t.Fatalf("acknowledging the delivered record after its file vanished: %v", err) + } + if got := w.Count(); got != 5 { + t.Fatalf("remaining count=%d, want 5", got) + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + reopened, err := New[uint64](dir, marshalU64, unmarshalU64, ackOpts()) + if err != nil { + t.Fatal(err) + } + defer func() { _ = reopened.Close() }() + got, reports := drainQueue(t, reopened.NewReader()) + if len(got) != 5 || got[0] != 240 || reports != 0 { + t.Fatalf("surviving segment: %v, reports=%d", got, reports) + } +} + +func emptyTornStore(t *testing.T) string { + t.Helper() + s, dir := newTestStore(t, 32, 0) + for i := 0; i < 4; i++ { + mustAppend(t, s, idxRec(i)) + } + last := s.active() + last.header(setCommitCursor(headerSize+last.size), setCommittedCount(last.written)) + if err := s.writeHeader(last); err != nil { + t.Fatal(err) + } + corruptData(t, s, 0, 0, unframeable) + if err := s.close(); err != nil { + t.Fatal(err) + } + tearHeader(t, dir, 1) + return dir +} + +func TestEmptyTornSegmentReplaysFollowingCommits(t *testing.T) { + dir := emptyTornStore(t) + for attempt := 0; attempt < 2; attempt++ { + reopened, err := openStore(dir, 32, 0, true, 0, 0) + if err != nil { + t.Fatal(err) + } + count := reopened.count() + if err := reopened.close(); err != nil { + t.Fatal(err) + } + if count != 2 { + t.Fatalf("open %d: recovered %d records, want the following 2 to replay", attempt, count) + } + } +} + +func TestPartialForceCommitKeepsCountsConsistent(t *testing.T) { + s, dir := newTestStore(t, 32, 0) + for i := 0; i < 6; i++ { + mustAppend(t, s, idxRec(i)) + } + reopenReadOnly(t, s, 1) + if err := s.forceCommitAll(); err == nil { + t.Fatal("force commit ignored a header write failure") + } + if s.count() != 4 || s.commitOff != s.files[1].base || s.headOff != s.commitOff { + t.Fatalf("partial commit left inconsistent state: count=%d commit=%d head=%d", + s.count(), s.commitOff, s.headOff) + } + repairHandle(t, s, 1) + if err := s.close(); err != nil { + t.Fatal(err) + } + reopened, err := openStore(dir, 32, 0, true, 0, 0) + if err != nil { + t.Fatal(err) + } + defer func() { _ = reopened.close() }() + if reopened.count() != 4 { + t.Fatalf("reopen disagrees with partial commit: count=%d", reopened.count()) + } +} + +func TestLatchedFailureRejectsAcknowledgementsBehindGaps(t *testing.T) { + w, _ := openAckQueue(t, ackOpts()) + addN(t, w, 3) + r := w.NewReader() + offs := reserveN(t, r, 3) + if err := r.Ack(offs[0]); err != nil { + t.Fatal(err) + } + breakHandle(t, w.st, 0) + if err := w.Sync(); !errors.Is(err, ErrIO) { + t.Fatalf("expected a durability failure, got %v", err) + } + repairHandle(t, w.st, 0) + for _, off := range []int64{offs[0], offs[2]} { + if err := r.Ack(off); !errors.Is(err, ErrIO) { + t.Errorf("Ack(%d) ignored the latched failure: %v", off, err) + } + if err := r.AckBatch(off); !errors.Is(err, ErrIO) { + t.Errorf("AckBatch(%d) ignored the latched failure: %v", off, err) + } + } + if _, err := w.AddBatch(nil); !errors.Is(err, ErrIO) { + t.Errorf("AddBatch ignored the latched failure: %v", err) + } +} + +func TestAppendReportsEvictionDurabilityFailure(t *testing.T) { + dir := t.TempDir() + s, err := openStore(dir, 32, 0, false, 1000, 3) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = s.close() }) + for i := 0; i < 6; i++ { + mustAppend(t, s, idxRec(i)) + } + breakHandle(t, s, 0) + if err := s.append(idxRec(6)); !errors.Is(err, ErrIO) { + t.Fatalf("append ignored an eviction flush failure: %v", err) + } + if s.count() != 6 { + t.Fatalf("failed append published a record: count=%d", s.count()) + } + _ = s.close() + reopened, err := openStore(dir, 32, 0, true, 0, 0) + if err != nil { + t.Fatal(err) + } + defer func() { _ = reopened.close() }() + if reopened.count() != 6 { + t.Fatalf("reopened count=%d, want 6", reopened.count()) + } +} + +func TestCommitReportsEvictionDurabilityFailure(t *testing.T) { + s, err := openStore(t.TempDir(), 32, 0, false, 1000, 3) + if err != nil { + t.Fatal(err) + } + defer func() { _ = s.close() }() + for i := 0; i < 8; i++ { + mustAppend(t, s, idxRec(i)) + } + // Reopening the first segment for this commit evicts the second, which is + // still dirty. Its failed flush must reach this operation's caller. + breakHandle(t, s, 1) + if err := s.commitTo(idxRecLen); !errors.Is(err, ErrIO) { + t.Fatalf("commit ignored an eviction flush failure: %v", err) + } + if s.commitOff != 0 { + t.Fatal("commit published after the durability failure") + } +} + +func TestSegmentNumberExhaustionDoesNotReorderRecords(t *testing.T) { + s, dir := newTestStore(t, 32, 0) + mustAppend(t, s, idxRec(1)) + mustAppend(t, s, idxRec(2)) + if err := s.close(); err != nil { + t.Fatal(err) + } + if err := os.Rename(s.filePath(1), s.filePath(^uint64(0))); err != nil { + t.Fatal(err) + } + reopened, err := openStore(dir, 32, 0, true, 0, 0) + if err != nil { + t.Fatal(err) + } + if err := reopened.append(idxRec(3)); err == nil { + t.Error("segment numbering wrapped around and inserted a new segment before the backlog") + } + if err := reopened.close(); err != nil { + t.Fatal(err) + } + again, err := openStore(dir, 32, 0, true, 0, 0) + if err != nil { + t.Fatal(err) + } + defer func() { _ = again.close() }() + got, _ := drainRecovering(t, again) + if len(got) != 2 || got[0] != 1 || got[1] != 2 { + t.Fatalf("numbering overflow changed the persisted backlog: %v", got) + } +} + +func TestCreateDoesNotOverwriteAnExistingSegment(t *testing.T) { + s, _ := newTestStore(t, 32, 0) + mustAppend(t, s, idxRec(1)) + mustAppend(t, s, idxRec(2)) + path := s.filePath(s.nextNum) + original := []byte("an existing file must not be truncated by create") + if err := os.WriteFile(path, original, 0o644); err != nil { + t.Fatal(err) + } + if err := s.append(idxRec(3)); !errors.Is(err, os.ErrExist) { + t.Errorf("append overwrote an existing path: %v", err) + } + got, err := os.ReadFile(path) + if err != nil || !bytes.Equal(got, original) { + t.Fatalf("existing file changed: %q, %v", got, err) + } +} + +func TestBatchPanicWakesReadersForPublishedPrefix(t *testing.T) { + for _, opts := range []Options{{}, {NoSync: true}, {SyncEvery: 2}} { + opts.SegmentSize = 4096 + w, err := New[int](t.TempDir(), func(dst []byte, v int) ([]byte, error) { + if v == 3 { + panic("codec") + } + return append(dst, make([]byte, 3000)...), nil + }, func([]byte) (int, error) { return 1, nil }, opts) + if err != nil { + t.Fatal(err) + } + // This is the same channel a blocked Reserve/Take/Follow waits on. + waiter := make(chan struct{}) + w.notify = waiter + func() { + defer func() { + if recover() == nil { + t.Error("codec did not panic") + } + }() + _, _ = w.AddBatch([]int{1, 2, 3}) + }() + select { + case <-waiter: + default: + t.Errorf("%+v: a published prefix left blocked readers asleep", opts) + } + want := 1 + if opts.SyncEvery == 2 { + want = 2 + } + if got := w.Count(); got != want { + t.Errorf("published %d records, want %d", got, want) + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + } +} + +func TestDecodeFailurePreservesLastDeliveryMetrics(t *testing.T) { + m, _ := bytesCodec() + w, err := New[[]byte](t.TempDir(), m, func(b []byte) ([]byte, error) { + if len(b) > 1 { + return nil, errors.New("decode") + } + return b, nil + }, Options{NoSync: true, StampRecords: true}) + if err != nil { + t.Fatal(err) + } + defer func() { _ = w.Close() }() + if _, err := w.AddBatch([][]byte{{1}, {2, 3}}); err != nil { + t.Fatal(err) + } + r := w.NewReader() + if _, ok, err := r.TryTake(); !ok || err != nil { + t.Fatalf("read: %v, %v", ok, err) + } + bytes, age := r.LastBytes(), r.LastAge() + if _, _, err := r.TryTake(); !errors.Is(err, ErrCodec) { + t.Fatal(err) + } + if r.LastBytes() != bytes || r.LastAge() != age { + t.Fatalf("failed decode replaced last delivery metrics: bytes %d -> %d, age %v -> %v", + bytes, r.LastBytes(), age, r.LastAge()) + } +} + +func TestFailedCommitCanBeRetried(t *testing.T) { + w, dir := openAckQueue(t, ackOpts()) + addN(t, w, 3) + r := w.NewReader() + offs := reserveN(t, r, 3) + reopenReadOnly(t, w.st, 0) + for i := 0; i < 2; i++ { + if err := r.AckBatch(offs...); err == nil { + t.Fatalf("attempt %d reported success without writing a header", i) + } + if w.Count() != 3 || w.st.commitOff != 0 || w.Stats().Committed != 0 { + t.Fatalf("failed commit changed retirement: %+v", w.Stats()) + } + } + repairHandle(t, w.st, 0) + if err := r.AckBatch(offs...); err != nil { + t.Fatal(err) + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + reopened, err := New[uint64](dir, marshalU64, unmarshalU64, ackOpts()) + if err != nil { + t.Fatal(err) + } + defer func() { _ = reopened.Close() }() + if reopened.Count() != 0 || !reopened.Empty() { + t.Fatalf("successful retry was not persisted: %+v", reopened.Stats()) + } +} + +func TestCommitRetainsOnlyPublishedPrefixAfterFailure(t *testing.T) { + s, dir := newTestStore(t, 32, 0) + for i := 0; i < 6; i++ { + mustAppend(t, s, idxRec(i)) + } + middle := s.files[1] + reopenReadOnly(t, s, 1) + if err := s.commitTo(s.writeOff); err == nil { + t.Fatal("commit ignored an unwritable middle header") + } + if s.commitOff != middle.base || s.count() != 4 || s.nCommittedTotal != 2 { + t.Fatalf("failed commit retired an unpublished suffix: %+v", s.stats()) + } + if err := s.close(); err != nil { + t.Fatal(err) + } + reopened, err := openStore(dir, 32, 0, true, 0, 0) + if err != nil { + t.Fatal(err) + } + defer func() { _ = reopened.close() }() + got, reports := drainRecovering(t, reopened) + if reports != 0 || len(got) != 4 || got[0] != 2 || got[3] != 5 { + t.Fatalf("failed commit changed the replayable suffix: %v, reports=%d", got, reports) + } +} + +func TestCommitCorruptionDoesNotPassItsTarget(t *testing.T) { + w, _ := openAckQueue(t, ackOpts()) + addN(t, w, 3) + r := w.NewReader() + offs := reserveN(t, r, 3) + corruptData(t, w.st, 0, 0, unframeable) + _ = r.Commit(offs[0]) + if w.st.commitOff > offs[0] || w.Count() < 2 { + t.Fatalf("damage in a prefix commit retired later reservations: %+v", w.Stats()) + } +} + +func TestCommitCorruptionWakesBlockedReaders(t *testing.T) { + w, _ := openAckQueue(t, ackOpts()) + addN(t, w, 3) + r := w.NewReader() + offs := reserveN(t, r, 3) + waiter := make(chan struct{}) + w.notify = waiter // a reader waiting after every record was reserved + corruptData(t, w.st, 0, 0, unframeable) + if err := r.Commit(offs[2]); err != nil { + t.Fatal(err) + } + if w.st.pendingCorrupt != 1 { + t.Fatalf("expected a commit-time corruption report, got %d", w.st.pendingCorrupt) + } + select { + case <-waiter: + default: + t.Fatal("a corruption report was queued without waking blocked readers") + } + if _, _, _, err := r.TryReserve(); !errors.Is(err, ErrCorrupt) { + t.Fatalf("pending report: %v", err) + } + if !w.Empty() { + t.Fatal("one corruption report should leave the queue empty") + } +} + +func TestReclaimedFileBehindPinnedPrefixIsNotLive(t *testing.T) { + s, _ := newTestStore(t, 32, 0) + for i := 0; i < 8; i++ { + mustAppend(t, s, idxRec(i)) + } + pinned, removed := s.files[0], s.files[1] + pinned.pins++ + if err := s.commitTo(s.writeOff); err != nil { + t.Fatal(err) + } + pinned.pins-- + if _, err := os.Stat(removed.path); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("expected reclaimed file: %v", err) + } + if s.live(removed) { + t.Fatal("reclaimed file is still considered live behind a pinned prefix") + } +} diff --git a/correctness_unix_test.go b/correctness_unix_test.go new file mode 100644 index 0000000..ce4cbdd --- /dev/null +++ b/correctness_unix_test.go @@ -0,0 +1,113 @@ +// SPDX-License-Identifier: MIT + +//go:build unix + +package diskqueue + +import ( + "bytes" + "os" + "testing" +) + +func TestRecoveryRequiresCommitReconciliationToPersist(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("root ignores file permissions") + } + s, dir := newTestStore(t, 32, 0) + for i := 0; i < 6; i++ { + mustAppend(t, s, idxRec(i)) + } + middle := s.files[1] + middle.header(setCommitCursor(headerSize+middle.size), setCommittedCount(middle.written)) + if err := s.writeHeader(middle); err != nil { + t.Fatal(err) + } + if err := s.close(); err != nil { + t.Fatal(err) + } + if err := os.Chmod(middle.path, 0o444); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.Chmod(middle.path, 0o644) }) + reopened, err := openStore(dir, 32, 0, true, 0, 0) + if err == nil { + _ = reopened.close() + t.Fatal("open ignored a failed repair of stale commit metadata") + } + if err := os.Chmod(middle.path, 0o644); err != nil { + t.Fatal(err) + } + reopened, err = openStore(dir, 32, 0, true, 0, 0) + if err != nil { + t.Fatal(err) + } + if reopened.count() != 6 { + t.Fatalf("recovered count=%d, want 6", reopened.count()) + } + if err := reopened.close(); err != nil { + t.Fatal(err) + } +} + +func TestPartialHeaderWriteRestoresPublishedImage(t *testing.T) { + if !loadIsolate(t) { + return + } + s, dir := newTestStore(t, 4096, 0) + mustAppend(t, s, idxRec(1)) + before := append([]byte(nil), s.active().hdr...) + // Stage the record before lowering the limit: only its publishing header + // write should be short. This is the same publication used by Add/AddBatch. + if err := s.stagePending(idxRec(2)); err != nil { + t.Fatal(err) + } + s.takeSpan() + restore := loadCapFileSize(t, 32) + err := s.publishSpan() + restore() + if err == nil { + t.Fatal("expected the publishing header write to be short") + } + if s.count() != 1 || s.stagedBytes() != 0 { + t.Fatalf("failed publication left in-memory state: %+v", s.stats()) + } + h, err := s.readHeader(1) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(h, before) { + t.Error("short write left a torn header instead of the previous published image") + } + if err := s.close(); err != nil { + t.Fatal(err) + } + reopened, err := openStore(dir, 4096, 0, true, 0, 0) + if err != nil { + t.Fatal(err) + } + _, off, ok, err := reopened.takeHead() + if err != nil || !ok { + t.Fatalf("read after failed publication: %v, %v", ok, err) + } + // Exercise a partial commit write too, using a header loaded during open + // rather than one created in this session. + restore = loadCapFileSize(t, 32) + err = reopened.commitTo(off) + restore() + if err == nil || reopened.count() != 1 || reopened.commitOff != 0 { + t.Fatalf("short commit write advanced retirement: err=%v stats=%+v", err, reopened.stats()) + } + if err := reopened.close(); err != nil { + t.Fatal(err) + } + again, err := openStore(dir, 4096, 0, true, 0, 0) + if err != nil { + t.Fatal(err) + } + defer func() { _ = again.close() }() + got, reports := drainRecovering(t, again) + if reports != 0 || len(got) != 1 || got[0] != 1 { + t.Fatalf("recovery resurrected a rejected append: %v, reports=%d", got, reports) + } +} diff --git a/cover_committo_test.go b/cover_committo_test.go index 199a462..6f93a68 100644 --- a/cover_committo_test.go +++ b/cover_committo_test.go @@ -247,14 +247,12 @@ func TestCommitToOpenFailureIsRetriableAndDeletesNothing(t *testing.T) { // cache looked exactly like one that did. // // The failure is a pwrite, not an fsync — class (2) — so it must be returned -// without poisoning the store, and what the walk already committed stands (the -// cursor only ever moved forward over records that were really there). +// without poisoning the store. Only headers successfully published before the +// failure may advance retirement; this failure is in the very first segment. func TestCommitToPersistFailureOnLeavingSegment(t *testing.T) { s, _ := newTestStore(t, 512, 0) committoFillSegments(t, s, 2) total := s.count() - firstSeg := s.files[0] - boundary := s.files[1].base // where the walk crosses out of segment 1 // Reads keep working, so the walk crosses the whole of segment 1; the header // write it does on the way out is what fails. @@ -270,19 +268,19 @@ func TestCommitToPersistFailureOnLeavingSegment(t *testing.T) { if errors.Is(err, ErrCorrupt) { t.Fatalf("a failed pwrite is not corruption: %v", err) } - if s.commitOff != boundary { - t.Fatalf("commitOff=%d, want %d: the walk stops where the header write failed", - s.commitOff, boundary) + if s.commitOff != 0 { + t.Fatalf("commitOff=%d, want 0: the first header write failed", s.commitOff) } - if want := total - firstSeg.written; s.count() != want { - t.Fatalf("count=%d, want %d: what was committed before the failure stands", s.count(), want) + if s.count() != total { + t.Fatalf("count=%d, want %d: no retirement was published", s.count(), total) } if st := s.stats(); st.Corruptions != 0 { t.Fatalf("corruptions=%d: a write failure is not damage", st.Corruptions) } - // The store is still usable: the rest of the backlog commits normally, now - // that the walk starts in a segment whose handle is healthy. + // Once the write failure clears, the same commit retries the unpublished + // segment instead of reporting success over a header it never persisted. + repairHandle(t, s, 0) if err := s.commitTo(s.writeOffset()); err != nil { t.Fatalf("second commitTo: %v", err) } diff --git a/cover_loadfile_test.go b/cover_loadfile_test.go index 040ccc5..557d1ba 100644 --- a/cover_loadfile_test.go +++ b/cover_loadfile_test.go @@ -611,15 +611,9 @@ func TestLoadFileCommitCursorPastSegmentClamped(t *testing.T) { } } -// TestAbortedCreateScanFailuresKeepTheCorruptionVerdict covers abortedCreate's two -// failure arms, which decide whether a dropped segment is COUNTED as loss. -// -// Neither is reachable through loadFile — a segment that cannot be opened has -// already failed readHeader, and the scan is bounded by the length os.Stat -// reported a moment earlier — so they are exercised directly. Both must answer in -// the conservative direction: silence is only ever bought by proof, because an -// under-reported loss is a segment that went missing with nothing said. -func TestAbortedCreateScanFailuresKeepTheCorruptionVerdict(t *testing.T) { +// An open/read failure must be reported separately from the verdict on a file's +// contents. EOF still proves an all-zero file if every surviving byte was read. +func TestAbortedCreateScanFailuresAreReported(t *testing.T) { dir := t.TempDir() s, err := openStore(dir, loadfileSegSize, 0, false, 0, 0) if err != nil { @@ -629,9 +623,8 @@ func TestAbortedCreateScanFailuresKeepTheCorruptionVerdict(t *testing.T) { zeroHdr := make([]byte, headerSize) full := int64(headerSize + loadfileSegSize) - // The scan cannot even open the file: nothing was proved, so the segment keeps - // the verdict its unreadable header earned. - if s.abortedCreate(404, full, zeroHdr) { + // The scan cannot open the file: recovery must fail without deleting it. + if aborted, err := s.abortedCreate(404, full, zeroHdr); aborted || err == nil { t.Fatal("a segment that could not be opened was reclaimed as an aborted create") } @@ -644,7 +637,7 @@ func TestAbortedCreateScanFailuresKeepTheCorruptionVerdict(t *testing.T) { } } writeSeg(900, make([]byte, headerSize+16)) - if !s.abortedCreate(900, full, zeroHdr) { + if aborted, err := s.abortedCreate(900, full, zeroHdr); !aborted || err != nil { t.Fatal("a wholly zero file that ended early was not recognised as an aborted create") } @@ -653,7 +646,7 @@ func TestAbortedCreateScanFailuresKeepTheCorruptionVerdict(t *testing.T) { late := make([]byte, headerSize+16) late[len(late)-1] = 1 writeSeg(901, late) - if s.abortedCreate(901, full, zeroHdr) { + if aborted, err := s.abortedCreate(901, full, zeroHdr); aborted || err != nil { t.Fatal("a file with data in it was reclaimed as an aborted create") } @@ -662,7 +655,7 @@ func TestAbortedCreateScanFailuresKeepTheCorruptionVerdict(t *testing.T) { writeSeg(902, make([]byte, full)) nonZeroHdr := make([]byte, headerSize) nonZeroHdr[headerSize-1] = 1 - if s.abortedCreate(902, full, nonZeroHdr) { + if aborted, err := s.abortedCreate(902, full, nonZeroHdr); aborted || err != nil { t.Fatal("a segment carrying a header was reclaimed as an aborted create") } } diff --git a/diskqueue.go b/diskqueue.go index 28a4592..3287072 100644 --- a/diskqueue.go +++ b/diskqueue.go @@ -83,9 +83,10 @@ var ( // another — already holds the directory's advisory lock. ErrLocked = errors.New("diskqueue: directory already in use") // ErrIO wraps a durability failure that the queue cannot recover from in - // place: an fsync that failed. The kernel reports such an error once and then - // drops the dirty pages, so a later fsync may well succeed with the data - // already gone — rather than report a durability it does not have, the queue + // place: a failed sync or an unrepairable partial header write. The kernel can + // report a writeback error once and then drop the dirty pages, so a later + // fsync may succeed with the data already gone. Rather than report a durability + // it does not have, the queue // latches the failure and every subsequent Add, commit and Sync returns it // (wrapping the original errno). Close it and reopen to continue. ErrIO = errors.New("diskqueue: durability failure") @@ -262,11 +263,8 @@ type Stats struct { // read — this is the number an operator alerts on, and the Lost* fields above // say how much each event cost. // - // It counts EVENTS, not damaged regions, and one region can produce more than - // one: a segment the read path quarantines is quarantined again when the commit - // walk later crosses it, so a single unframable record can read as Corruptions=2 - // with LostSegments=1. The byte and record figures are booked once; this one - // tracks the reports actually handed to consumers. + // A deferred retirement of damage already reported by a read does not book + // another event when an acknowledgement later advances across it. Corruptions uint64 } @@ -483,7 +481,7 @@ func (w *Queue[T]) putBuf(mb *marshalBuf) { // pwrite) leaves the queue untouched: the error means the item is not in it. // // A durability failure is the one AMBIGUOUS answer. If the error wraps ErrIO the -// item may or may not be in the log, and the two arms are not distinguishable +// item may or may not be in the log, and these cases are not distinguishable // from the error: a failed HEADER fsync leaves the record real to everything // short of a power loss (its bytes and the header publishing them both reached // the page cache), while a failed DATA fsync publishes nothing at all. Either way @@ -578,12 +576,15 @@ func (w *Queue[T]) AddWait(ctx context.Context, data T) error { // AddBatch appends items in order, amortizing the lock and — under the per-op // policy — the fsyncs across the whole batch: the records' bytes go down // first, one data fsync covers them all, then one header write and one header -// fsync publish them together (per segment crossed). Every published record is -// durable when AddBatch returns. +// fsync publish them together (per segment crossed). With the default policy, +// every published record is durable on a nil error. Deferred sync policies keep +// their usual power-loss window. // // It returns how many leading items are in the queue. n < len(items) comes // with the error that stopped the batch (ErrFull, a marshal error, an I/O -// failure); the first n items are placed and durable regardless. +// failure). The first n items were published; a final header-sync failure leaves +// them visible in the page cache but returns ErrIO, so their durability is +// uncertain, just as with Add. // // Unlike Add and AddWait, which marshal BEFORE taking the lock, AddBatch calls // MarshalFunc with the queue mutex held. A codec that panics there is therefore a @@ -599,12 +600,13 @@ func (w *Queue[T]) AddBatch(items []T) (int, error) { return 0, ErrClosed } before := w.st.writeOffset() - n, err := w.addBatchLocked(items) - w.dropOversizedScratch() - if w.st.writeOffset() != before { - w.signal() - } - return n, err + defer func() { + w.dropOversizedScratch() + if w.st.writeOffset() != before { + w.signal() // a codec panic can follow an already-published segment + } + }() + return w.addBatchLocked(items) } // addBatchLocked stages every record and publishes one span per segment crossed, @@ -642,6 +644,9 @@ func (w *Queue[T]) addBatchLocked(items []T) (int, error) { if w.closed { return 0, ErrClosed // Close ran while the quiesce had the lock released } + if st.ioErr != nil { + return 0, st.ioErr + } published := 0 for _, it := range items { b, err := w.marshal(w.stampInto(w.scratch[:0]), it) @@ -867,6 +872,9 @@ func (w *Queue[T]) leadFlushLocked() error { err = datasync(f) w.mu.Lock() } + if err == nil { + err = st.ioErr // a concurrent commit/eviction may have failed its flush + } switch { case err != nil: // The span's bytes may be gone and nothing was published: discard @@ -888,6 +896,9 @@ func (w *Queue[T]) leadFlushLocked() error { // and the store is poisoned. err = st.failIO(herr) } + if err == nil { + err = st.ioErr + } } // dirty deliberately stays set: a commit may have rewritten the // header while the lock was away, and clearing it here would @@ -921,6 +932,7 @@ func (w *Queue[T]) leadFlushLocked() error { } st.flushing = false w.signalFlushDrained() + w.signalSpace() // rollback freed staged capacity, or a latch must wake AddWait return mine } @@ -1015,6 +1027,16 @@ func (w *Queue[T]) signalSpace() { } } +// signalProgress wakes producers after a consume operation and readers when a +// commit discovered corruption after the read cursor had already reached the +// tail. Pending reports are readable work too. The caller holds w.mu. +func (w *Queue[T]) signalProgress() { + w.signalSpace() + if w.st.pendingCorrupt > 0 { + w.signal() + } +} + // dropOversizedScratch releases the marshal buffer once it has grown past the // segment geometry — the size of the largest ORDINARY record, and therefore of // anything the steady state ever needs. Oversized records are admitted (into @@ -1134,7 +1156,7 @@ func (w *Queue[T]) Sync() error { return w.syncOffLock() } // than resetting, which is what keeps them honest under concurrent producers. // // The dirty set is listed under the lock but pinned and opened a chunk at a -// time (flushChunkSize, one below MaxOpenFiles), because a pinned file is +// time (flushChunkSize, two below MaxOpenFiles), because a pinned file is // exactly the one evictOpen may not close: pinning the whole set at once — as // this did — took MaxOpenFiles out of force for the duration of the flush and, // since nothing re-evicts on the way out, after it as well. Under NoSync that @@ -1155,6 +1177,7 @@ func (w *Queue[T]) syncOffLock() error { defer w.syncMu.Unlock() w.mu.Lock() defer w.mu.Unlock() + defer w.signalSpace() // a flush failure must wake producers to observe the latch if w.closed { return ErrClosed } @@ -1206,6 +1229,9 @@ func (w *Queue[T]) syncOffLock() error { // counted here — unless an interleaved flushBatch already zeroed the gauges // (the epoch moved), in which case they now describe only what accumulated // since it and the stale snapshot must not be subtracted from that. + if err := st.syncDir(); err != nil { + return err + } if st.flushEpoch == snapEpoch { st.unsynced = max(st.unsynced-snapOps, 0) st.unsyncedBytes = max(st.unsyncedBytes-snapBytes, 0) diff --git a/doc.go b/doc.go index 7569fe8..0708002 100644 --- a/doc.go +++ b/doc.go @@ -101,8 +101,10 @@ that arrives while another Add's fsync is in flight joins the next flush span, so one data fsync and one header fsync cover every record that joined (group commit). Each Add still returns only once its own record is durable — the sharing changes the cost, not the contract. [Queue.AddBatch] amortizes the -same way across a batch from one goroutine, and returns how many leading items -were placed, each durable, when it stops early. +same way across a batch from one goroutine and returns how many leading items +were published. Their durability follows the selected sync policy; a final +header-sync failure returns [ErrIO] with the published count and uncertain +durability, as with Add. How often that fsync happens is the main throughput knob: @@ -114,10 +116,11 @@ How often that fsync happens is the main throughput knob: checksum on read. - [Options].SyncInterval adds a wall-clock backstop, so an idle queue's last writes become durable on a timer rather than waiting for N more operations. - - [Options].NoSync never syncs. Data still survives a process crash through the + - [Options].NoSync disables per-operation syncs. Data survives a process crash through the page cache, but not a power loss. -[Queue.Sync] flushes on demand and [Queue.Close] always flushes. A flush's +[Queue.Sync] flushes on demand and [Queue.Close] always flushes, including the +directory entries that name the segments on supported platforms. A flush's per-file fdatasyncs run WITHOUT the queue's lock held — pinned against eviction and reclamation instead — so the SyncInterval backstop over a deep unsynced backlog does not stall every concurrent Add and read for the duration of the @@ -135,6 +138,10 @@ it does not have, the queue latches the failure: every subsequent Add, commit an Sync returns [ErrIO] wrapping the original errno. Reads keep working, so a poisoned queue can still be drained. Close and reopen to continue. +A partial header overwrite is restored and flushed before returning a retriable +write error. If that restoration cannot be completed, the queue also latches +[ErrIO]: the published extent is uncertain until recovery. + # Crash recovery and corruption Two rules govern everything the recovery path does: @@ -154,6 +161,10 @@ that segment with it, because the record boundaries behind it are gone too. A genuine I/O error is not damage at all: nothing is dropped and the cursor stays put, since the bytes may still be there next time. +Retiring damage never retires an earlier outstanding reservation. The damaged +extent waits in the acknowledgement ledger until the earlier work is acknowledged; +that later retirement does not report the corruption again. + Reopening reads no records — one 64-byte header pread per segment, and the header is the single source of truth for the write cursor, the resume point and the record count. Two kinds of segment are excepted, both already known to be @@ -168,6 +179,10 @@ vouch for themselves, and the one thing that cannot be reconstructed is the commit position, so that segment and everything after it replays (at-least-once). One corruption event is reported for it. +Required header repairs are flushed before opening completes, even under NoSync. +Following commit headers are reset before a salvaged header is replaced, so an interrupted repair +preserves the replay decision even when the salvaged segment is empty. + On reopen the read cursor resets to the persisted commit cursor, so uncommitted items replay: the crash guarantee is at-least-once. [Queue.Rewind] does the same thing without reopening, returning in-flight (reserved but uncommitted) records diff --git a/docs/guide.md b/docs/guide.md index 602df9c..d6967ee 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -113,7 +113,7 @@ processing (at-least-once), use `Reserve`/`Ack`. | `Add(v T) error` | Append an item. Returns `ErrFull` at `MaxSegments` or `MaxBytes` (transient), `ErrRecordTooLarge` if it can never fit under `MaxBytes` (permanent). A record too big for one segment gets a segment of its own. Serialization runs before the lock, and under the default policy concurrent `Add`s share their fsyncs (group commit) while each record stays individually durable. | | `AddSized(v T) (int64, error)` | `Add`, also reporting the framed size the record occupies on disk — the unit `MaxBytes` and the byte gauges count (`Reader.LastBytes` is the consuming side of the same number). | | `AddWait(ctx, v T) error` | `Add` with backpressure: blocks while the queue is full, until a commit frees capacity or `ctx` is done. `ErrRecordTooLarge` still returns immediately. | -| `AddBatch(items []T) (int, error)` | Append items in order with the per-record cost amortized over the whole batch: one header write per segment crossed on every policy, plus (under per-op durability) one data fsync and one header fsync. Returns how many leading items were placed — on error the first `n` are in and durable, the rest are not. | +| `AddBatch(items []T) (int, error)` | Append items in order with header writes and syncs shared across each published span. Returns how many leading items were published. Durability follows the selected policy; a final header-sync failure returns `ErrIO` with the published count and uncertain durability. | | `NewReader() *Reader[T]` | Create a Reader to consume items (one per consuming goroutine). | | `Empty() bool` | Whether anything is available to read. | | `Count() int` | Number of items added but not yet committed. | @@ -225,6 +225,17 @@ Reading is where losses are reported, so a consumer that never reads never learn of them; `Stats()` is readable at any time, including after `Close`. +Damage is retired through the acknowledgement ledger when earlier reservations +are outstanding, including reservations in the damaged segment itself. An `Ack` +that closes the gap retires the reported damage without reporting it again and +without retiring records appended afterward. + +Recovery must flush corrected cursors and counts before opening succeeds, even +under `NoSync`. A required repair that cannot be written fails the open so it can +be retried. For a torn header, later commit headers are reset before the salvaged +header is replaced; an interrupted repair therefore preserves the replay decision even +when no records survived in the salvaged segment. + ```go for { v, ok, err := r.TryTake() @@ -259,6 +270,8 @@ and the queue is left in a state you can reason about. `ErrRecordTooLarge` or a failed `pwrite` leaves the queue byte-for-byte as it was: the item is not in it, and retrying is well defined. A segment that cannot be created is unlinked again, so a failed `Add` leaves no stub behind. + A partial header overwrite is restored to the preceding published image; if + that restoration fails, the queue latches `ErrIO` and placement is uncertain. - **A failed `fsync` poisons the queue.** Linux reports a writeback error *once* and then discards the dirty pages, so a second `fsync` can return success over data that is already gone. Rather than claim a durability it cannot deliver, the @@ -270,8 +283,9 @@ and the queue is left in a state you can reason about. still on disk and uncommitted records replay. - **Commits report failure.** `Take`/`TryTake` return the item *and* a non-nil error when the read succeeded but its commit did not reach disk: the item is - yours to process, and it will be delivered again after a reopen. `Commit` - returns the same error directly. + yours to process and may replay after a reopen. A failed header write leaves + the affected segment's retirement unchanged and can be retried; a failed final + sync leaves durability uncertain. `Commit` returns the same error directly. - **Iterators report through `Err()`.** An `iter.Seq` cannot carry an error, so `Drain`/`Follow` stop and record it: check `r.Err()` after the loop. Nil means the iteration ended because the queue was drained, the context was cancelled, or @@ -366,7 +380,8 @@ and clears as the consumer drains. `SyncEvery` trades a bounded power-loss window for throughput: with `SyncEvery: N` the fsync cost is amortized over N writes/commits, so up to the last N unsynced operations can be lost on power loss (they still survive a process crash, and a -torn tail is caught by the per-record checksum). `Close` and `Sync` always flush. +torn tail is caught by the per-record checksum). `Close` and `Sync` always flush, +including segment directory entries on supported platforms, even under `NoSync`. `Stats().UnsyncedBytes` reports that window as a number — zero under the default per-op policy, climbing under `NoSync` or `SyncEvery > 1` until a flush. If it keeps climbing, the `SyncInterval` backstop is not keeping up. diff --git a/header.go b/header.go index 26996fc..3991db6 100644 --- a/header.go +++ b/header.go @@ -52,8 +52,12 @@ type dataFile struct { path string f *os.File // open handle, or nil when not currently open hdr []byte // resident copy of the 64-byte header (page 0) - base int64 // global offset of this file's first data byte - size int64 // bytes of records written into the data region (excludes header) + // The last complete header image written or read from disk. Header modifiers + // change hdr before publication; this image lets a short overwrite restore + // exactly the bytes it changed without resurrecting an unpublished append. + diskHeader [headerSize]byte + base int64 // global offset of this file's first data byte + size int64 // bytes of records written into the data region (excludes header) // capacity is how many record bytes this segment's data region can hold. It is // the store's segmentSize for an ordinary segment, and exactly one record's // framed length for an oversized one — which is what keeps "records never span @@ -93,6 +97,9 @@ type dataFile struct { // re-extends the file — otherwise the next open reads the zero fill back as // records. truncated bool + // A salvaged header has no trustworthy commit position, even when no record + // survived. Recovery must rewind the following segments before replacing it. + resetCommit bool } // Header layout (little-endian): [0:8] magic, [8:16] commit cursor, [16:24] write diff --git a/reader.go b/reader.go index 46fcddc..453c0fd 100644 --- a/reader.go +++ b/reader.go @@ -98,7 +98,7 @@ func (r *Reader[T]) TryReserve() (T, bool, int64, error) { var zero T r.w.mu.Lock() defer r.w.mu.Unlock() - defer r.w.signalSpace() // a quarantine on the read path can advance the commit cursor + defer r.w.signalProgress() // a quarantine on the read path can advance the commit cursor if r.w.closed { return zero, false, 0, ErrClosed } @@ -113,13 +113,13 @@ func (r *Reader[T]) TryReserve() (T, bool, int64, error) { // TryTake returns and commits the front item; ok is false when empty. // // A non-nil error with ok true means the item was read but its commit could not -// be persisted: the item is yours, and it will be delivered again after a reopen -// (the commit, not the read, is what is missing). +// be persisted: the item is yours and may replay after a reopen. A failed header +// write leaves it uncommitted; a failed final sync leaves durability uncertain. func (r *Reader[T]) TryTake() (T, bool, error) { var zero T r.w.mu.Lock() defer r.w.mu.Unlock() - defer r.w.signalSpace() // the commit may have freed capacity a producer waits on + defer r.w.signalProgress() // the commit may have freed capacity a producer waits on if r.w.closed { return zero, false, ErrClosed } @@ -136,7 +136,7 @@ func (r *Reader[T]) Reserve(ctx context.Context) (T, bool, int64, error) { var zero T r.w.mu.Lock() defer r.w.mu.Unlock() - defer r.w.signalSpace() // a quarantine on the read path can advance the commit cursor + defer r.w.signalProgress() // a quarantine on the read path can advance the commit cursor for { if r.w.closed { return zero, false, 0, ErrClosed @@ -157,12 +157,12 @@ func (r *Reader[T]) Reserve(ctx context.Context) (T, bool, int64, error) { // Take blocks until an item is available (or ctx is done) and returns + commits // it. As with TryTake, a non-nil error alongside ok true means the item was read -// but its commit did not reach disk, so it replays after a reopen. +// but its commit failed, so it may replay after a reopen. func (r *Reader[T]) Take(ctx context.Context) (T, bool, error) { var zero T r.w.mu.Lock() defer r.w.mu.Unlock() - defer r.w.signalSpace() // the commit may have freed capacity a producer waits on + defer r.w.signalProgress() // the commit may have freed capacity a producer waits on for { if r.w.closed { return zero, false, ErrClosed @@ -192,7 +192,7 @@ func (r *Reader[T]) Take(ctx context.Context) (T, bool, error) { func (r *Reader[T]) Commit(offset int64) error { r.w.mu.Lock() defer r.w.mu.Unlock() - defer r.w.signalSpace() // the commit may have freed capacity a producer waits on + defer r.w.signalProgress() // the commit may have freed capacity a producer waits on if r.w.closed { return ErrClosed } @@ -230,7 +230,7 @@ func (r *Reader[T]) Commit(offset int64) error { func (r *Reader[T]) Ack(offset int64) error { r.w.mu.Lock() defer r.w.mu.Unlock() - defer r.w.signalSpace() // the commit may have freed capacity a producer waits on + defer r.w.signalProgress() // the commit may have freed capacity a producer waits on if r.w.closed { return ErrClosed } @@ -279,11 +279,14 @@ func (r *Reader[T]) LastAge() time.Duration { return r.lastAge } func (r *Reader[T]) AckBatch(offsets ...int64) error { r.w.mu.Lock() defer r.w.mu.Unlock() - defer r.w.signalSpace() // the commit may have freed capacity a producer waits on + defer r.w.signalProgress() // the commit may have freed capacity a producer waits on if r.w.closed { return ErrClosed } st := r.w.st + if st.ioErr != nil { + return st.ioErr + } st.dropRetired() head := st.headOffset() invalid := false @@ -336,7 +339,7 @@ func (r *Reader[T]) AckBatch(offsets ...int64) error { func (r *Reader[T]) Skip() (bool, error) { r.w.mu.Lock() defer r.w.mu.Unlock() - defer r.w.signalSpace() // the commit may have freed capacity a producer waits on + defer r.w.signalProgress() // the commit may have freed capacity a producer waits on if r.w.closed { return false, ErrClosed } @@ -404,7 +407,7 @@ func (r *Reader[T]) next(ctx context.Context, follow bool, end int64) (T, bool, w := r.w w.mu.Lock() defer w.mu.Unlock() - defer w.signalSpace() // iterators commit as they read; producers may be waiting + defer w.signalProgress() // iterators commit as they read; producers may be waiting for { if w.closed { return zero, false, nil @@ -494,7 +497,8 @@ func (r *Reader[T]) read() (T, int64, bool, error) { // start, which a quarantine en route may have moved past (see trackReserve). // The stamp counts: it is part of the payload, so LastBytes stays the exact // unit MaxBytes and AddSized report. - r.lastBytes = r.w.st.lastFrameEnd - r.w.st.lastFrameAt + lastBytes := r.w.st.lastFrameEnd - r.w.st.lastFrameAt + var lastAge time.Duration defer r.trimScratch() // on the codec exit too; see TryPeek body := r.scratch if r.w.stamp { @@ -507,7 +511,7 @@ func (r *Reader[T]) read() (T, int64, bool, error) { // and never a panic. return zero, 0, false, fmt.Errorf("%w: %w", ErrCodec, errShortStamp) } - r.lastAge = stampAge(body) + lastAge = stampAge(body) body = body[stampLen:] } v, err := r.w.unmarshal(body) @@ -521,6 +525,7 @@ func (r *Reader[T]) read() (T, int64, bool, error) { return zero, 0, false, fmt.Errorf("%w: %w", ErrCodec, err) } delivered = true + r.lastBytes, r.lastAge = lastBytes, lastAge return v, off, true, nil } @@ -576,7 +581,7 @@ func (r *Reader[T]) trimScratch() { func (r *Reader[T]) Requeue() (bool, error) { r.w.mu.Lock() defer r.w.mu.Unlock() - defer r.w.signalSpace() // the head commit may free capacity a producer waits on + defer r.w.signalProgress() // the head commit may free capacity a producer waits on if r.w.closed { return false, ErrClosed } diff --git a/recovery.go b/recovery.go index 2dc5d0a..076efc0 100644 --- a/recovery.go +++ b/recovery.go @@ -103,7 +103,7 @@ func (s *store) load() error { // Commit cursor: the first file whose commit cursor is short of its end. s.commitOff = s.writeOff for i, df := range s.files { - if commitCurs[i] < headerSize+df.size { + if df.resetCommit || commitCurs[i] < headerSize+df.size { s.commitOff = df.base + (commitCurs[i] - headerSize) break } @@ -151,7 +151,12 @@ func (s *store) load() error { // // On a healthy open the guard never fires — every field already agrees — so // reopening still costs one 64-byte pread per segment and no writes. - for _, df := range s.files { + // Repair from the tail backwards: an empty salvaged segment is the only + // evidence that following commits must replay. Replace that torn header only + // after the later commit headers have been reset, so an interrupted repair + // cannot erase the witness before its consequences reach disk. + for i := len(s.files) - 1; i >= 0; i-- { + df := s.files[i] // The commit cursor this segment must publish, reconciled against the // recovered GLOBAL cursor exactly as the count pass above reconciles the // counts: everything below it is committed, everything above it is not, and @@ -168,17 +173,13 @@ func (s *store) load() error { df.commitCursor() == cc && df.committedCount() == df.committed { continue } - // Fatal only for a TRUNCATED segment, whose republish must land or the same - // loss is re-booked on every open. For the reconciliation-only population - // this is best-effort, like the re-extension below: a segment that can be - // read but not written (a read-only file, a full disk) must not take its - // healthy siblings down with it, and leaving that header uncorrected is - // exactly the state every release before this one shipped. - fatal := df.truncated + // Cursor/count reconciliation is part of recovery, not best-effort + // maintenance: leaving a stale committed header can silently retire + // replayed work once its preceding segment is reclaimed. if err := s.ensureOpen(df); err != nil { - if !fatal { - continue - } + return err + } + if err := faultPoint("recovery.writeHeader"); err != nil { return err } df.header( @@ -188,18 +189,16 @@ func (s *store) load() error { setCommittedCount(df.committed), ) if err := s.writeHeader(df); err != nil { - if !fatal { - continue - } return err } - if !s.noSync { - if err := s.flushFile(df); err != nil { - if !fatal { - continue - } - return err - } + // Recovery repairs must survive before the file is re-extended or a + // torn header's replay witness is replaced, even with NoSync. That option + // defers normal operations; it cannot defer this repair ordering. + if err := faultPoint("recovery.syncHeader"); err != nil { + return s.failIO(err) + } + if err := s.flushFile(df); err != nil { + return err } // Restore the geometry too, and only after the clamped cursor is durable. // Re-extended, the segment is an ordinary partly-filled file whose tail is @@ -305,8 +304,14 @@ func (s *store) loadFile(num uint64, base int64, tail bool) (*dataFile, int64, e // zeros rather than a zero-length one. Counted as damage it reports a whole // segment's worth of loss — 8 MiB at the default geometry — for a file that // never held a byte, which is what an operator sees after a power cut. - if tail && s.abortedCreate(num, size, h) { - return nil, 0, s.removeFile(num) + if tail { + aborted, err := s.abortedCreate(num, size, h) + if err != nil { + return nil, 0, err + } + if aborted { + return nil, 0, s.removeFile(num) + } } return nil, 0, s.dropSegment(num, size, false) case !th.headerChecksumOK(): @@ -328,6 +333,7 @@ func (s *store) loadFile(num uint64, base int64, tail bool) (*dataFile, int64, e return nil, 0, serr } if ok { + copy(df.diskHeader[:], h) return df, cc, nil } } @@ -380,6 +386,7 @@ func (s *store) loadFile(num uint64, base int64, tail bool) (*dataFile, int64, e } df := &dataFile{num: num, path: path, hdr: h, base: base, size: w - headerSize, capacity: capacity, truncated: truncated} + copy(df.diskHeader[:], h) df.written = max(th.writtenCount(), 0) if lostTail { // The header counts records the file no longer holds, so Count() would @@ -398,6 +405,12 @@ func (s *store) loadFile(num uint64, base int64, tail bool) (*dataFile, int64, e // for the repair below to republish and the next append to land behind. // Fail the open — the segment is untouched and a retry once the error // clears recovers everything. + if df.f != nil { + _ = df.f.Close() // read-only walk; no unpublished writes + df.f = nil + s.untrackOpen(df) + } + s.dropBlock() return nil, 0, werr } { @@ -574,6 +587,7 @@ func (s *store) salvageTornHeader(num uint64, path string, base, size int64) (*d // durably and re-extends the file — the same aftercare a truncated segment // gets, for the same reason: the next open must find a header it can believe. df.truncated = true + df.resetCommit = true s.corruptions++ s.pendingCorrupt++ @@ -635,23 +649,21 @@ func (s *store) countVerified(df *dataFile) (int64, int64, error) { // time at the end of the sequence, and the number advances only once the // create has succeeded. A zeroed segment with a higher-numbered sibling is // therefore something else, and stays corruption. -// - A scan that cannot finish answers false, leaving the segment with the -// verdict its unbelievable header already earned. Failing to look licenses -// nothing here — the unlink was licensed already, and what is at stake is -// only whether it is counted as loss. +// - An I/O failure during the scan fails recovery. A partial read cannot +// establish whether the file is an aborted create or contains data. // // Writing the header before reserving the blocks would not settle the same // ambiguity: the reservation is metadata and can reach the disk while the // header's bytes are still in the page cache, leaving exactly this file again. -func (s *store) abortedCreate(num uint64, size int64, h []byte) bool { +func (s *store) abortedCreate(num uint64, size int64, h []byte) (bool, error) { for _, b := range h { if b != 0 { - return false + return false, nil } } f, err := os.Open(s.filePath(num)) if err != nil { - return false + return false, err } defer func() { _ = f.Close() }() // read-only handle: nothing to lose on close // The walk stops at the first nonzero byte, so a segment that holds records @@ -659,20 +671,26 @@ func (s *store) abortedCreate(num uint64, size int64, h []byte) bool { // on an open that found a header of zeros. buf := make([]byte, 64<<10) for off := int64(headerSize); off < size; { + if err := faultPoint("recovery.readZero"); err != nil { + return false, err + } n, rerr := f.ReadAt(buf[:min(int64(len(buf)), size-off)], off) + if rerr != nil && !isShortRead(rerr) { + return false, rerr + } for _, b := range buf[:n] { if b != 0 { - return false + return false, nil } } if rerr != nil { // The file ended before the length os.Stat reported: everything that is // there has been read, and all of it was zero. - return isShortRead(rerr) + return true, nil } off += int64(n) } - return true + return true, nil } // dropSegment unlinks a segment that cannot be read and books the loss: as diff --git a/reserve.go b/reserve.go index e5145cd..6779a5a 100644 --- a/reserve.go +++ b/reserve.go @@ -36,6 +36,11 @@ type reservation struct { start int64 end int64 acked bool + // A corrupt span covers a segment's abandoned tail. written snapshots the + // segment's published count at detection, so later appends to the same file + // cannot be retired along with the damage. + corrupt bool + written int64 } // trackReserve records the boundary of the record a Reserve just handed over. @@ -139,6 +144,9 @@ func (s *store) ackedRunEnd() int64 { // part way leaves the ledger describing what actually happened and the next ack // picks up from there. func (s *store) ackTo(end int64) (bool, error) { + if s.ioErr != nil { + return true, s.ioErr + } s.dropRetired() if end <= s.commitOff { return true, nil // already retired, here or by another consume path @@ -157,11 +165,18 @@ func (s *store) ackTo(end int64) (bool, error) { // reservation commits nothing and is not an error. Shared by ackTo, AckBatch and // retireFrame. func (s *store) commitAckedRun() error { + return s.commitAckedRunDurable(true) +} + +func (s *store) commitAckedRunDurable(durable bool) error { + if s.ioErr != nil { + return s.ioErr + } target := s.ackedRunEnd() if target <= s.commitOff { return nil // acknowledged, but a gap ahead still blocks the cursor } - err := s.commitTo(target) + err := s.commitDurable(target, durable) s.dropRetired() // keep the ledger in step with however far the commit actually got return err } @@ -180,7 +195,11 @@ func (s *store) commitAckedRun() error { // Appending keeps the ledger sorted: the frame was the read head, so its end is // past every outstanding reservation's. func (s *store) retireFrame(start, end int64) error { + return s.retireFrameDurable(start, end, true) +} + +func (s *store) retireFrameDurable(start, end int64, durable bool) error { s.dropRetired() s.reserved = append(s.reserved, reservation{start: start, end: end, acked: true}) - return s.commitAckedRun() + return s.commitAckedRunDurable(durable) } diff --git a/store.go b/store.go index 877b874..757cf17 100644 --- a/store.go +++ b/store.go @@ -214,13 +214,15 @@ func openStore(dir string, segmentSize int64, maxSegments int, noSync bool, sync return s, nil } -// failIO latches an unrecoverable durability failure. Only fsync failures land -// here: the kernel reports a writeback error once and then drops the dirty pages, -// so a retry can report success with the data already gone. Rather than claim a +// failIO latches an unrecoverable durability failure: a failed fsync, or a partial +// header overwrite whose previous image could not be restored. The kernel can +// report a writeback error once and then drop the dirty pages, so a retry can +// report success with the data already gone. Rather than claim a // durability it cannot deliver, the store remembers the first such failure and // every later append/commit/sync repeats it; the caller's recourse is to close -// and reopen. Write and open failures are *not* latched — they leave the store -// consistent and are safe to retry. +// and reopen. Ordinary write and open failures are not latched: they leave the +// store consistent and are safe to retry. A failed header restoration cannot make +// that promise, so it takes the same latch rather than guessing what was published. func (s *store) failIO(err error) error { if err == nil { return nil @@ -334,31 +336,25 @@ func (s *store) evictOpen(keep *dataFile) { } } -// live reports whether df is still one of the store's segments — for a caller -// holding a pointer it collected under an earlier acquisition of the lock, like -// the chunked off-lock flush. Reclamation only ever removes a leading PREFIX of -// s.files (see dropCommitted), and segment bases only ever increase, so a file -// has been unlinked exactly when its base sits below the live range: no -// per-file "dropped" flag to keep in step with every removal path. A file -// dropCommitted could not unlink stays in s.files, which only makes this answer -// conservatively "live". +// live checks membership for a pointer retained across lock holds. A pinned or +// un-unlinkable file can survive while later committed files are reclaimed, so +// comparing against the first live base alone cannot establish membership. func (s *store) live(df *dataFile) bool { - return len(s.files) > 0 && df.base >= s.files[0].base + i := sort.Search(len(s.files), func(i int) bool { return s.files[i].num >= df.num }) + return i < len(s.files) && s.files[i] == df } // flushChunkSize is how many files an off-lock flush (Queue.syncOffLock) may // hold open at once; 0 means no limit, matching an unbounded maxOpenFiles. // -// One below the cap, because the chunk's files are pinned and the active file -// is never evictable: leaving that one slot free is what keeps nOpen at or -// under maxOpenFiles for the whole flush rather than one over it whenever the -// active file is not itself in the chunk. openStore floors a nonzero -// maxOpenFiles at 3, so a chunk always holds at least two files. +// Leave slots for both the active file and a concurrent reader's newly opened +// file: neither is evictable during ensureOpen. openStore floors a nonzero cap +// at 3, so even the smallest cap permits one pinned file. func (s *store) flushChunkSize() int { if s.maxOpenFiles <= 0 { return 0 } - return max(s.maxOpenFiles-1, 1) + return max(s.maxOpenFiles-2, 1) } // batched reports whether the sync policy defers fsync to a periodic flush @@ -405,6 +401,12 @@ func (s *store) flushBatch() error { if errs != nil { return errs } + if s.ioErr != nil { + return s.ioErr // reopening a dirty file may have failed an eviction flush + } + if err := s.syncDir(); err != nil { + return err + } s.unsynced, s.unsyncedBytes = 0, 0 s.flushEpoch++ return nil @@ -419,8 +421,11 @@ func (s *store) filePath(num uint64) string { // to trip over, and returns cleanly: nothing was published, so the store is // exactly where it was. func (s *store) createFile(num uint64, base, capacity int64) (*dataFile, error) { + if num == 0 { + return nil, errors.New("diskqueue: segment number space exhausted") + } path := s.filePath(num) - f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0o644) + f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_EXCL, 0o644) if err != nil { return nil, err } @@ -487,15 +492,52 @@ func (s *store) writeHeader(df *dataFile) error { return err } } + if s.ioErr != nil { + return s.ioErr // ensureOpen can latch a dirty victim's failed flush + } if _, err := df.f.WriteAt(df.hdr, 0); err != nil { + // A closed os.File rejects the call before issuing any write. Other + // failures can have overwritten a prefix even when WriteAt reports n=0, + // so read back the header rather than trusting that count. + if !errors.Is(err, os.ErrClosed) { + if rerr := s.restoreHeader(df); rerr != nil { + return errors.Join(err, s.failIO(rerr)) + } + } return err } + copy(df.diskHeader[:], df.hdr) s.nHeaderWrites++ df.dirty = true df.writeSeq++ return nil } +// restoreHeader restores the changed prefix of a failed header overwrite. It +// flushes the restoration before reporting a retriable write failure, so recovery +// cannot mistake the torn image for a reason to salvage an unpublished append. +func (s *store) restoreHeader(df *dataFile) error { + var current [headerSize]byte + if _, err := df.f.ReadAt(current[:], 0); err != nil { + return fmt.Errorf("checking failed header write: %w", err) + } + n := headerSize + for n > 0 && current[n-1] == df.diskHeader[n-1] { + n-- + } + if n == 0 { + return nil // the failed write left the published header intact + } + df.dirty = true + df.writeSeq++ + // Restore only bytes that changed. A file-size limit can reject a complete + // header overwrite while still allowing this smaller repair to succeed. + if _, err := df.f.WriteAt(df.diskHeader[:n], 0); err != nil { + return fmt.Errorf("restoring partial header: %w", err) + } + return datasync(df.f) +} + // flushFile fsyncs df if it has unsynced writes, then marks it clean. No-op for // an already-clean file; a dirty file whose handle was evicted is reopened, // because dirty means "has unsynced bytes", not "is open". @@ -512,6 +554,9 @@ func (s *store) flushFile(df *dataFile) error { return err } } + if s.ioErr != nil { + return s.ioErr + } if err := datasync(df.f); err != nil { return s.failIO(err) } @@ -565,6 +610,9 @@ func (s *store) appendRecord(payload []byte, force bool) error { if err := s.ensureOpen(af); err != nil { return err } + if s.ioErr != nil { + return s.ioErr // ensureOpen/cycle may have failed a dirty eviction + } if err := faultPoint("append.writeRecord"); err != nil { return err @@ -718,6 +766,9 @@ func (s *store) stagePending(payload []byte) error { if err := s.ensureOpen(af); err != nil { return err } + if s.ioErr != nil { + return s.ioErr + } if err := faultPoint("append.writeRecord"); err != nil { return err } @@ -829,6 +880,9 @@ func (s *store) cycle(need int64, force bool) error { s.diskBytes += headerSize + df.capacity s.trackOpen(df) // Persist the new (and removed) entries before records land in the file. + if s.ioErr != nil { + return s.ioErr + } if !s.noSync { if err := s.syncDir(); err != nil { return err @@ -1003,33 +1057,13 @@ func (s *store) takeHead() ([]byte, int64, bool, error) { s.corruptions++ s.headOff = next cerr := fmt.Errorf("%w: record checksum", ErrCorrupt) - // Retire the record that was just reported destroyed. Nothing else ever - // would: every consume op returns on this error without committing, so a - // damaged record left at the head keeps Count() above zero for good, pins its - // segment against reclamation, and is re-read and re-reported on every reopen - // — one ErrCorrupt and one LostRecords per restart, forever. - // - // Only when the commit cursor stood exactly here, though. With records - // reserved behind it the cursor may not jump: that would retire work no - // consumer has acknowledged. Those cases normally heal on their own, because - // the eventual Commit walks across this record like any other — its framing - // is intact, only the payload failed its checksum. - // - // The residue, stated so nobody rediscovers it as a bug: a Reserve/Commit - // consumer that NEVER acknowledges past the damage leaves Count() at the - // number of records it is still holding, with Empty() true, and pins their - // segment for the session. That is the same thing an unacknowledged reserve - // does to any record, damaged or not — the store keeps no per-record - // reservation state, so it cannot tell "handed out and outstanding" from - // "handed out and dropped", and guessing would retire live work. Rewind or a - // Commit past the damage clears it. - if head == s.commitOff && s.ioErr == nil { - // Unflushed (see commitDurable) and deliberately unreported: the record - // is already booked lost and the cursor has already moved, so ErrCorrupt - // alone is the whole truth the caller needs. Joining a write failure on - // here would turn one damaged record into a second, unrelated error on a - // path whose contract is "damage was dropped, go round again". - _ = s.commitDurable(next, false) + // Keep the dropped frame in the same contiguous retirement ledger as + // Skip. Otherwise an Ack before the damage leaves an unfillable hole, + // while a prefix commit would retire another worker's reservation. + // Reads do not spend a SyncEvery tick, but a failed header write must + // still reach the caller; the ledger retains the retirement for retry. + if err := s.retireFrameDurable(head, next, false); err != nil { + return nil, 0, false, errors.Join(cerr, err) } return nil, 0, false, cerr } @@ -1152,9 +1186,12 @@ func (s *store) forceCommitAll() error { return err } if abandoned := df.written - df.committed; abandoned > 0 { + s.nCommitted += abandoned s.nCommittedTotal += uint64(abandoned) } df.committed = df.written + s.commitOff = max(s.commitOff, df.base+df.size) + s.headOff = max(s.headOff, s.commitOff) } // Square the global counter outright instead of summing the per-file moves // above, because the two differ in exactly the case this arm exists to answer: @@ -1172,10 +1209,10 @@ func (s *store) forceCommitAll() error { return nil } -// skipCorruptSegment abandons the rest of the segment holding off: it advances -// the read cursor past the segment, and — when the commit cursor is already -// within that segment (the auto-committing read path) — force-commits the -// abandoned tail so it is reclaimed and never replayed. +// skipCorruptSegment abandons the rest of the segment holding off. It advances +// the read cursor and retires the damage immediately only when commitOff has +// reached off. Otherwise the abandoned extent enters the acknowledgement ledger, +// so earlier reservations remain live, including ones in the same segment. // // This is the expensive half of the blast-radius rule, reached only when the // framing itself is gone: with the record boundaries unknown, there is nothing @@ -1199,12 +1236,13 @@ func (s *store) skipCorruptSegment(off int64) error { // not carry is replayed wholesale by the next open, while nCommittedTotal — a // lifetime counter no reopen undoes — has already counted those records, so // Stats().Committed would count them again as they are re-consumed. + head := s.headOff if err := s.forceCommitAll(); err != nil { return err } s.corruptions++ s.lostSegments++ - if lost := s.writeOff - s.headOff; lost > 0 { + if lost := s.writeOff - head; lost > 0 { s.lostBytes += uint64(lost) } s.headOff = s.writeOff @@ -1214,7 +1252,7 @@ func (s *store) skipCorruptSegment(off int64) error { return nil } end := df.base + df.size - if s.commitOff >= df.base { + if s.commitOff == off { // A segment whose file has vanished leaves no header to record the // quarantine in and nothing to replay from it, so publishFullCommit // succeeds and the in-memory advance below is all there is to do; @@ -1245,6 +1283,15 @@ func (s *store) skipCorruptSegment(off int64) error { if s.headOff < end { s.headOff = end } + if s.commitOff < end { + // Defer retirement behind earlier work, including a reservation in this + // same segment. Snapshot the extent/count so concurrent publication or a + // later append cannot be mistaken for part of this abandoned tail. + s.dropRetired() + s.reserved = append(s.reserved, reservation{ + start: off, end: end, acked: true, corrupt: true, written: df.written, + }) + } s.corruptions++ return nil } @@ -1283,7 +1330,8 @@ func (s *store) commitDurable(off int64, durable bool) error { perOp := durable && s.perOp() var cur *dataFile // file with header changes not yet written out var stop error - advanced := false + before := s.commitOff + reservedIndex := 0 var df *dataFile // carried across iterations; the walk is sequential for s.commitOff < off { // Re-resolve only when the cursor leaves the carried file: re-scanning the @@ -1300,17 +1348,58 @@ func (s *store) commitDurable(off int64, durable bool) error { if cur != nil && cur != df { // Leaving a file: write its final header now, while its handle is // certainly still open — opening the next segment can evict this one. - if err := s.persistCommit(cur, perOp); err != nil { + previous := cur + cur = nil + if err := s.persistCommit(previous, perOp); err != nil { stop = err break } - cur = nil } - if err := s.ensureOpen(df); err != nil { - stop = err // can't open the file to advance the cursor; replay later + // The ledger and commit walk both advance in order. Carry its position + // across records instead of searching the whole in-flight set each time. + for reservedIndex < len(s.reserved) && s.reserved[reservedIndex].end <= s.commitOff { + reservedIndex++ + } + var reserved *reservation + if reservedIndex < len(s.reserved) && s.reserved[reservedIndex].start == s.commitOff { + reserved = &s.reserved[reservedIndex] + } + if reserved != nil && reserved.corrupt { + if reserved.end > off { + break + } + // Its framing is known to be unusable. Retire exactly the extent + // already reported lost, without walking it or reporting it again. + abandoned := max(reserved.written-df.committed, 0) + s.commitOff = reserved.end + s.nCommitted += abandoned + s.nCommittedTotal += uint64(abandoned) + df.committed += abandoned + cur = df + continue + } + openErr := s.ensureOpen(df) + if s.ioErr != nil { + stop = s.ioErr + break + } + if openErr != nil && !errors.Is(openErr, os.ErrNotExist) { + stop = openErr // inaccessible is not corrupt: retry later break } - next, ok, err := s.frameEnd(df, s.commitOff) + var next int64 + var ok bool + var err error + switch { + case openErr == nil: + next, ok, err = s.frameEnd(df, s.commitOff) + case reserved != nil: + // The bytes vanished after delivery, but this reservation's boundary + // is still known. Acknowledging it must not retire its neighbours. + next, ok = reserved.end, true + default: + err = fmt.Errorf("%w: %w", ErrCorrupt, openErr) + } // The cursor may only ever move forward, and never past the tail: a decoded // length that says otherwise is corruption, not a destination. if ok && (next <= s.commitOff || next > s.writeOff) { @@ -1325,6 +1414,12 @@ func (s *store) commitDurable(off int64, durable bool) error { // applies. A cursor stuck on an unframable record would also stop all // reclamation, so the disk fills behind it. if errors.Is(err, ErrCorrupt) { + if off < df.base+df.size { + // A prefix commit is not permission to discard outstanding + // reservations or unread records beyond its target. + stop = err + break + } if serr := s.skipCorruptSegment(s.commitOff); serr != nil { stop = serr break @@ -1335,7 +1430,7 @@ func (s *store) commitDurable(off int64, durable bool) error { // climbs with nothing in any log is what breaks an operator's trust // in the number. s.pendingCorrupt++ - cur, advanced = nil, true // it persisted the header itself + cur = nil // it persisted the header itself continue } stop = err @@ -1357,14 +1452,13 @@ func (s *store) commitDurable(off int64, durable bool) error { // recomputed the 56-byte checksum for every record crossed and threw all but // the last away — 43% of the commit walk, with the queue mutex held. cur = df - advanced = true } if cur != nil { if err := s.persistCommit(cur, perOp); err != nil && stop == nil { stop = err } } - if !advanced { + if s.commitOff == before { return stop // nothing committed } if durable && s.batched() { @@ -1394,11 +1488,24 @@ func (s *store) commitDurable(off int64, durable bool) error { // The cursor is clamped to df's own extent, because this is also the call that // finalizes a file the walk has already left — and such a file is fully committed. func (s *store) persistCommit(df *dataFile, perOp bool) error { + prevCursor, prevCount := df.commitCursor(), df.committedCount() df.header( setCommitCursor(headerSize+min(s.commitOff-df.base, df.size)), setCommittedCount(df.committed), ) if err := s.writeHeader(df); err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil // a vanished segment has neither a header nor data to replay + } + // Only successfully published segments may remain retired. Keeping this + // advance would make an Ack retry a no-op and allow reclamation to unlink + // work whose commit never reached even the page cache. + rolledBack := df.committed - prevCount + s.commitOff = df.base + prevCursor - headerSize + s.nCommitted -= rolledBack + s.nCommittedTotal -= uint64(rolledBack) + df.committed = prevCount + df.header(setCommitCursor(prevCursor), setCommittedCount(prevCount)) return err } if perOp { @@ -1484,6 +1591,9 @@ func (s *store) syncDir() error { if s.dirFile == nil { return nil } + if err := faultPoint("sync.dir"); err != nil { + return s.failIO(err) + } err := fsyncDir(s.dirFile) if err == nil || dirSyncUnsupported(err) { return nil @@ -1515,8 +1625,12 @@ func (s *store) close() error { if err := df.f.Close(); err != nil && first == nil { first = err } + s.untrackOpen(df) df.f = nil } + if first == nil { + first = s.syncDir() + } if first == nil { // Every file reached stable storage, so the power-loss exposure is genuinely // zero — and Stats stays readable after Close, so it has to say so. A latched