From 312cd858dd37888dc8e8d9fc9a30c6ac73d99b41 Mon Sep 17 00:00:00 2001 From: bootjp Date: Tue, 27 Oct 2026 04:49:36 +0900 Subject: [PATCH 01/11] filesystem: expose journal accounting diagnostics Signed-off-by: bootjp --- internal/filesystem/operation_journal.go | 7 +++++-- internal/filesystem/operation_journal_exec.go | 6 +++++- internal/filesystem/operation_record.go | 19 +++++++++++-------- 3 files changed, 21 insertions(+), 11 deletions(-) diff --git a/internal/filesystem/operation_journal.go b/internal/filesystem/operation_journal.go index 26d345766..6c12439a4 100644 --- a/internal/filesystem/operation_journal.go +++ b/internal/filesystem/operation_journal.go @@ -449,8 +449,11 @@ func (j *OperationJournal) loadRecordAt(ctx context.Context, id OperationID, ts return operationRecord{}, false, errors.Wrap(err, "read operation record") } record, err := decodeOperationRecord(raw) - if err != nil || record.ID != id { - return operationRecord{}, false, errors.Wrap(ErrOperationRecordCorrupt, "invalid operation row") + if err != nil { + return operationRecord{}, false, errors.Wrap(err, "decode operation row") + } + if record.ID != id { + return operationRecord{}, false, errors.Wrap(ErrOperationRecordCorrupt, "operation row identity mismatch") } return record, true, nil } diff --git a/internal/filesystem/operation_journal_exec.go b/internal/filesystem/operation_journal_exec.go index f9c675d90..666cd4ff5 100644 --- a/internal/filesystem/operation_journal_exec.go +++ b/internal/filesystem/operation_journal_exec.go @@ -665,7 +665,11 @@ func reserveAckRetirement(session *operationSession, record operationRecord) err return errors.WithStack(ErrOperationRetired) } if session.RetainedCount == 0 || session.RetainedBytes < record.Reservation { - return errors.Wrap(ErrOperationRecordCorrupt, "ack reservation exceeds retained usage") + return errors.Wrapf(ErrOperationRecordCorrupt, + "ack reservation exceeds retained usage: sequence=%d generation=%d state=%d retained_count=%d retained_bytes=%d reservation=%d pending=%d reject_before=%d ack_gaps=%d", + record.ID.Sequence, record.Generation, record.State, session.RetainedCount, + session.RetainedBytes, record.Reservation, session.Pending, + session.RejectBefore, len(session.AckGaps)) } if err := advanceAckWatermark(session, record.ID.Sequence); err != nil { return err diff --git a/internal/filesystem/operation_record.go b/internal/filesystem/operation_record.go index 5a08aa245..bebb88f8f 100644 --- a/internal/filesystem/operation_record.go +++ b/internal/filesystem/operation_record.go @@ -119,18 +119,21 @@ func decodeOperationRecord(raw []byte) (operationRecord, error) { return operationRecord{}, errors.WithStack(ErrOperationRecordCorrupt) } record.CommitTS = binary.BigEndian.Uint64(raw[operationCommitTSOffset:]) - if !validDecodedOperationRecord(raw, record) { - return operationRecord{}, errors.WithStack(ErrOperationRecordCorrupt) + if raw[4] == operationRecordLegacyVersion && (record.RecordIsPrimary || len(record.FencedAttempts) != 0) { + return operationRecord{}, errors.Wrap(ErrOperationRecordCorrupt, "legacy operation row has primary metadata") + } + if record.State != OperationPending && record.CommitTS == 0 { + return operationRecord{}, errors.Wrap(ErrOperationRecordCorrupt, "terminal operation row has no commit timestamp") + } + if record.State != OperationState(raw[5]) { + return operationRecord{}, errors.Wrap(ErrOperationRecordCorrupt, "operation row state differs from header") + } + if err := validateOperationRecord(record); err != nil { + return operationRecord{}, errors.Wrap(err, "invalid decoded operation row") } return record, nil } -func validDecodedOperationRecord(raw []byte, record operationRecord) bool { - return (raw[4] != operationRecordLegacyVersion || (!record.RecordIsPrimary && len(record.FencedAttempts) == 0)) && - (record.State == OperationPending || record.CommitTS != 0) && - record.State == OperationState(raw[5]) && validateOperationRecord(record) == nil -} - func validOperationHeader(raw []byte) bool { return len(raw) >= operationHeaderBytes && len(raw) <= maxOperationRecordBytes && bytes.Equal(raw[:4], []byte("FSOP")) && From c5ef83d3bf8604175fa39bad87fdce5f7d57d662 Mon Sep 17 00:00:00 2001 From: bootjp Date: Tue, 27 Oct 2026 04:55:08 +0900 Subject: [PATCH 02/11] filesystem: cover journal accounting under sustained churn Signed-off-by: bootjp --- .../operation_journal_accounting_test.go | 50 +++++++++++++++++++ 1 file changed, 50 insertions(+) create mode 100644 internal/filesystem/operation_journal_accounting_test.go diff --git a/internal/filesystem/operation_journal_accounting_test.go b/internal/filesystem/operation_journal_accounting_test.go new file mode 100644 index 000000000..56bbcde39 --- /dev/null +++ b/internal/filesystem/operation_journal_accounting_test.go @@ -0,0 +1,50 @@ +package filesystem + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestOperationJournalRetainedAccountingAcrossSequenceBoundary(t *testing.T) { + ctx := context.Background() + journal, _, _ := testOperationJournal(t) + first := testOperationID(t, journal, 1) + const batchSize = 16 + const operations = 1280 + + for start := uint64(1); start <= operations; start += batchSize { + for sequence := start; sequence < start+batchSize; sequence++ { + id := OperationID{Filesystem: first.Filesystem, Session: first.Session, Sequence: sequence} + _, err := journal.Admit(ctx, id, []byte("write"), 128, 1024) + require.NoError(t, err, "admit sequence %d", sequence) + attempt, err := journal.BeginAttempt(ctx, id) + require.NoError(t, err, "begin sequence %d", sequence) + if sequence%3 == 0 { + attempt, err = journal.AdvanceGeneration(ctx, id, attempt.AttemptID) + require.NoError(t, err, "advance sequence %d", sequence) + } + _, err = journal.Publish(ctx, id, attempt.AttemptID, operationAttemptPlan{ + Intent: []byte("write"), Result: []byte("done"), + }) + require.NoError(t, err, "publish sequence %d", sequence) + } + for sequence := start + batchSize - 1; sequence >= start; sequence-- { + id := OperationID{Filesystem: first.Filesystem, Session: first.Session, Sequence: sequence} + require.NoError(t, journal.Acknowledge(ctx, id), "ack sequence %d", sequence) + } + } + + ts, err := journal.backend.Snapshot(ctx) + require.NoError(t, err) + session, found, err := journal.loadSessionAt(ctx, OperationSessionID{ + Filesystem: first.Filesystem, Session: first.Session, + }, ts) + require.NoError(t, err) + require.True(t, found) + require.Equal(t, uint64(operations), session.RejectBefore) + require.Zero(t, session.Pending) + require.Zero(t, session.RetainedCount) + require.Zero(t, session.RetainedBytes) +} From 5397917dd79c57c6e30845526ed2f849f41500a0 Mon Sep 17 00:00:00 2001 From: bootjp Date: Tue, 27 Oct 2026 04:57:31 +0900 Subject: [PATCH 03/11] filesystem: exercise retained accounting on Pebble Signed-off-by: bootjp --- .../operation_journal_accounting_test.go | 20 ++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/internal/filesystem/operation_journal_accounting_test.go b/internal/filesystem/operation_journal_accounting_test.go index 56bbcde39..10e11b5a7 100644 --- a/internal/filesystem/operation_journal_accounting_test.go +++ b/internal/filesystem/operation_journal_accounting_test.go @@ -4,12 +4,30 @@ import ( "context" "testing" + "github.com/bootjp/elastickv/store" "github.com/stretchr/testify/require" ) func TestOperationJournalRetainedAccountingAcrossSequenceBoundary(t *testing.T) { + t.Run("memory", func(t *testing.T) { + journal, _, _ := testOperationJournal(t) + assertOperationJournalRetainedAccountingAcrossSequenceBoundary(t, journal) + }) + t.Run("pebble", func(t *testing.T) { + st, err := store.NewPebbleStore(t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, st.Close()) }) + service, err := NewService(st, &testCoordinator{st: st}) + require.NoError(t, err) + journal, err := NewOperationJournal(service) + require.NoError(t, err) + assertOperationJournalRetainedAccountingAcrossSequenceBoundary(t, journal) + }) +} + +func assertOperationJournalRetainedAccountingAcrossSequenceBoundary(t *testing.T, journal *OperationJournal) { + t.Helper() ctx := context.Background() - journal, _, _ := testOperationJournal(t) first := testOperationID(t, journal, 1) const batchSize = 16 const operations = 1280 From 91214247b5de736c320d6bb5e7e9380ed586a50e Mon Sep 17 00:00:00 2001 From: bootjp Date: Tue, 27 Oct 2026 05:00:28 +0900 Subject: [PATCH 04/11] filesystem: retain session validation context Signed-off-by: bootjp --- internal/filesystem/operation_journal.go | 7 +++++-- internal/filesystem/operation_session.go | 8 ++++++-- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/internal/filesystem/operation_journal.go b/internal/filesystem/operation_journal.go index 6c12439a4..3dfe131ef 100644 --- a/internal/filesystem/operation_journal.go +++ b/internal/filesystem/operation_journal.go @@ -434,8 +434,11 @@ func (j *OperationJournal) loadSessionAt(ctx context.Context, id OperationSessio return operationSession{}, false, errors.Wrap(err, "read operation session") } session, err := decodeOperationSession(raw) - if err != nil || session.ID != id { - return operationSession{}, false, errors.Wrap(ErrOperationRecordCorrupt, "invalid operation session row") + if err != nil { + return operationSession{}, false, errors.Wrap(err, "decode operation session row") + } + if session.ID != id { + return operationSession{}, false, errors.Wrap(ErrOperationRecordCorrupt, "operation session identity mismatch") } return session, true, nil } diff --git a/internal/filesystem/operation_session.go b/internal/filesystem/operation_session.go index 0868e3131..460e82d81 100644 --- a/internal/filesystem/operation_session.go +++ b/internal/filesystem/operation_session.go @@ -85,10 +85,14 @@ func validateOperationSession(session operationSession) error { return errors.Wrap(ErrOperationRecordCorrupt, "invalid operation session identity") } if !validOperationSessionUsage(session) { - return errors.Wrap(ErrOperationRecordCorrupt, "invalid operation session usage") + return errors.Wrapf(ErrOperationRecordCorrupt, + "invalid operation session usage: pending=%d retained_count=%d retained_bytes=%d max_count=%d max_bytes=%d", + session.Pending, session.RetainedCount, session.RetainedBytes, session.MaxCount, session.MaxBytes) } if !validOperationAckGaps(session) { - return errors.Wrap(ErrOperationRecordCorrupt, "invalid operation session acknowledgement gaps") + return errors.Wrapf(ErrOperationRecordCorrupt, + "invalid operation session acknowledgement gaps: reject_before=%d ack_gaps=%d", + session.RejectBefore, len(session.AckGaps)) } return nil } From 76c7d89ffee4096c772d18c200142d7cb0d24808 Mon Sep 17 00:00:00 2001 From: bootjp Date: Tue, 27 Oct 2026 05:12:29 +0900 Subject: [PATCH 05/11] filesystem: identify pending journal accounting mismatch Signed-off-by: bootjp --- internal/filesystem/operation_journal_exec.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/internal/filesystem/operation_journal_exec.go b/internal/filesystem/operation_journal_exec.go index 666cd4ff5..aeccc7fd6 100644 --- a/internal/filesystem/operation_journal_exec.go +++ b/internal/filesystem/operation_journal_exec.go @@ -519,7 +519,10 @@ func validatePendingSnapshot(record operationRecord, session operationSession, f return errors.WithStack(ErrOperationGeneration) } if record.ID.Session != session.ID.Session || session.Pending == 0 { - return errors.WithStack(ErrOperationRecordCorrupt) + return errors.Wrapf(ErrOperationRecordCorrupt, + "pending operation lacks session accounting: sequence=%d generation=%d pending=%d retained_count=%d retained_bytes=%d reservation=%d", + record.ID.Sequence, record.Generation, session.Pending, session.RetainedCount, + session.RetainedBytes, record.Reservation) } return nil } From c8fc5ff9f0c401184907a28989761d9bfdbab018 Mon Sep 17 00:00:00 2001 From: bootjp Date: Tue, 27 Oct 2026 14:55:10 +0900 Subject: [PATCH 06/11] filesystem: verify concurrent ACK accounting Signed-off-by: bootjp --- .../operation_journal_accounting_test.go | 81 +++++++++++++++++++ 1 file changed, 81 insertions(+) diff --git a/internal/filesystem/operation_journal_accounting_test.go b/internal/filesystem/operation_journal_accounting_test.go index 10e11b5a7..7e8bc60fd 100644 --- a/internal/filesystem/operation_journal_accounting_test.go +++ b/internal/filesystem/operation_journal_accounting_test.go @@ -2,6 +2,7 @@ package filesystem import ( "context" + "sync" "testing" "github.com/bootjp/elastickv/store" @@ -25,6 +26,86 @@ func TestOperationJournalRetainedAccountingAcrossSequenceBoundary(t *testing.T) }) } +func TestOperationJournalConcurrentAcknowledgementsPreserveAccounting(t *testing.T) { + t.Run("memory", func(t *testing.T) { + journal, _, _ := testOperationJournal(t) + assertConcurrentAcknowledgementsPreserveAccounting(t, journal) + }) + t.Run("pebble", func(t *testing.T) { + st, err := store.NewPebbleStore(t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, st.Close()) }) + service, err := NewService(st, &testCoordinator{st: st}) + require.NoError(t, err) + journal, err := NewOperationJournal(service) + require.NoError(t, err) + assertConcurrentAcknowledgementsPreserveAccounting(t, journal) + }) +} + +func assertConcurrentAcknowledgementsPreserveAccounting(t *testing.T, journal *OperationJournal) { + t.Helper() + ctx := context.Background() + first := testOperationID(t, journal, 1) + ids := []OperationID{first, {Filesystem: first.Filesystem, Session: first.Session, Sequence: 2}} + for _, id := range ids { + _, err := journal.Admit(ctx, id, []byte("write"), 128, 1024) + require.NoError(t, err) + attempt, err := journal.BeginAttempt(ctx, id) + require.NoError(t, err) + _, err = journal.Publish(ctx, id, attempt.AttemptID, operationAttemptPlan{ + Intent: []byte("write"), Result: []byte("done"), + }) + require.NoError(t, err) + } + + sessionID := OperationSessionID{Filesystem: first.Filesystem, Session: first.Session} + backend := &synchronizeSessionReadsBackend{ + operationJournalBackend: journal.backend, + sessionKey: sessionID.key(), + both: make(chan struct{}), + } + concurrent := newOperationJournalWithBackend(backend) + type outcome struct { + id OperationID + err error + } + results := make(chan outcome, len(ids)) + var workers sync.WaitGroup + for _, id := range ids { + workers.Add(1) + go func() { + defer workers.Done() + results <- outcome{id: id, err: concurrent.acknowledgeOnce(ctx, id)} + }() + } + workers.Wait() + close(results) + + succeeded, conflicted := 0, 0 + for result := range results { + if result.err == nil { + succeeded++ + continue + } + require.True(t, IsWriteConflict(result.err), "ack sequence %d: %v", result.id.Sequence, result.err) + conflicted++ + require.NoError(t, journal.Acknowledge(ctx, result.id)) + } + require.Equal(t, 1, succeeded) + require.Equal(t, 1, conflicted) + + ts, err := journal.backend.Snapshot(ctx) + require.NoError(t, err) + session, found, err := journal.loadSessionAt(ctx, sessionID, ts) + require.NoError(t, err) + require.True(t, found) + require.Equal(t, uint64(2), session.RejectBefore) + require.Zero(t, session.Pending) + require.Zero(t, session.RetainedCount) + require.Zero(t, session.RetainedBytes) +} + func assertOperationJournalRetainedAccountingAcrossSequenceBoundary(t *testing.T, journal *OperationJournal) { t.Helper() ctx := context.Background() From d8be5d465b879a5659aff2678188db3c156905e7 Mon Sep 17 00:00:00 2001 From: bootjp Date: Tue, 27 Oct 2026 14:57:24 +0900 Subject: [PATCH 07/11] filesystem: identify invalid operation record fields Signed-off-by: bootjp --- internal/filesystem/operation_record.go | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/internal/filesystem/operation_record.go b/internal/filesystem/operation_record.go index bebb88f8f..59d25dfa2 100644 --- a/internal/filesystem/operation_record.go +++ b/internal/filesystem/operation_record.go @@ -142,8 +142,24 @@ func validOperationHeader(raw []byte) bool { } func validateOperationRecord(record operationRecord) error { - if !validOperationCore(record) || !validOperationEffects(record) || !validOperationState(record) || !validFencedAttempts(record) { - return errors.WithStack(ErrOperationRecordCorrupt) + if !validOperationCore(record) { + expectedReservation := 2*(uint64(len(record.Intent))+uint64(record.ResultLimit)+uint64(record.EffectLimit)) + + operationHeaderBytes + operationReservationOverhead + + uint64(len(record.FencedAttempts))*operationFenceReservation + return errors.Wrapf(ErrOperationRecordCorrupt, + "invalid operation record core: sequence=%d state=%d generation=%d reservation=%d expected_reservation=%d intent_bytes=%d result_bytes=%d result_limit=%d effect_limit=%d fenced_attempts=%d fingerprint_matches=%t", + record.ID.Sequence, record.State, record.Generation, record.Reservation, expectedReservation, + len(record.Intent), len(record.Result), record.ResultLimit, record.EffectLimit, + len(record.FencedAttempts), record.Fingerprint == sha256.Sum256(record.Intent)) + } + if !validOperationEffects(record) { + return errors.Wrap(ErrOperationRecordCorrupt, "invalid operation record effects") + } + if !validOperationState(record) { + return errors.Wrap(ErrOperationRecordCorrupt, "invalid operation record state") + } + if !validFencedAttempts(record) { + return errors.Wrap(ErrOperationRecordCorrupt, "invalid operation record fenced attempts") } return nil } From 2a34a3521f8b4a1b8239a585e58947371af1d8e1 Mon Sep 17 00:00:00 2001 From: bootjp Date: Tue, 27 Oct 2026 14:58:42 +0900 Subject: [PATCH 08/11] filesystem: keep operation decoder checks focused Signed-off-by: bootjp --- internal/filesystem/operation_record.go | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/internal/filesystem/operation_record.go b/internal/filesystem/operation_record.go index 59d25dfa2..0f3a769c1 100644 --- a/internal/filesystem/operation_record.go +++ b/internal/filesystem/operation_record.go @@ -119,19 +119,26 @@ func decodeOperationRecord(raw []byte) (operationRecord, error) { return operationRecord{}, errors.WithStack(ErrOperationRecordCorrupt) } record.CommitTS = binary.BigEndian.Uint64(raw[operationCommitTSOffset:]) + if err := validateDecodedOperationRecord(raw, record); err != nil { + return operationRecord{}, err + } + return record, nil +} + +func validateDecodedOperationRecord(raw []byte, record operationRecord) error { if raw[4] == operationRecordLegacyVersion && (record.RecordIsPrimary || len(record.FencedAttempts) != 0) { - return operationRecord{}, errors.Wrap(ErrOperationRecordCorrupt, "legacy operation row has primary metadata") + return errors.Wrap(ErrOperationRecordCorrupt, "legacy operation row has primary metadata") } if record.State != OperationPending && record.CommitTS == 0 { - return operationRecord{}, errors.Wrap(ErrOperationRecordCorrupt, "terminal operation row has no commit timestamp") + return errors.Wrap(ErrOperationRecordCorrupt, "terminal operation row has no commit timestamp") } if record.State != OperationState(raw[5]) { - return operationRecord{}, errors.Wrap(ErrOperationRecordCorrupt, "operation row state differs from header") + return errors.Wrap(ErrOperationRecordCorrupt, "operation row state differs from header") } if err := validateOperationRecord(record); err != nil { - return operationRecord{}, errors.Wrap(err, "invalid decoded operation row") + return errors.Wrap(err, "invalid decoded operation row") } - return record, nil + return nil } func validOperationHeader(raw []byte) bool { From 9af7de1968ea9d55dde0ed33a50d4d1f87f8abd4 Mon Sep 17 00:00:00 2001 From: bootjp Date: Tue, 27 Oct 2026 15:00:34 +0900 Subject: [PATCH 09/11] filesystem: stress concurrent ACK retirement Signed-off-by: bootjp --- .../operation_journal_accounting_test.go | 74 +++++++++++++++++++ 1 file changed, 74 insertions(+) diff --git a/internal/filesystem/operation_journal_accounting_test.go b/internal/filesystem/operation_journal_accounting_test.go index 7e8bc60fd..daa5b0cf5 100644 --- a/internal/filesystem/operation_journal_accounting_test.go +++ b/internal/filesystem/operation_journal_accounting_test.go @@ -3,6 +3,7 @@ package filesystem import ( "context" "sync" + "sync/atomic" "testing" "github.com/bootjp/elastickv/store" @@ -26,6 +27,79 @@ func TestOperationJournalRetainedAccountingAcrossSequenceBoundary(t *testing.T) }) } +func TestOperationJournalConcurrentAcknowledgementChurn(t *testing.T) { + t.Run("memory", func(t *testing.T) { + journal, _, _ := testOperationJournal(t) + assertConcurrentAcknowledgementChurn(t, journal) + }) + t.Run("pebble", func(t *testing.T) { + st, err := store.NewPebbleStore(t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, st.Close()) }) + service, err := NewService(st, &testCoordinator{st: st}) + require.NoError(t, err) + journal, err := NewOperationJournal(service) + require.NoError(t, err) + assertConcurrentAcknowledgementChurn(t, journal) + }) +} + +func assertConcurrentAcknowledgementChurn(t *testing.T, journal *OperationJournal) { + t.Helper() + ctx := context.Background() + first := testOperationID(t, journal, 1) + const operations = uint64(256) + for sequence := uint64(1); sequence <= operations; sequence++ { + id := OperationID{Filesystem: first.Filesystem, Session: first.Session, Sequence: sequence} + _, err := journal.Admit(ctx, id, []byte("write"), 128, 1024) + require.NoError(t, err) + attempt, err := journal.BeginAttempt(ctx, id) + require.NoError(t, err) + _, err = journal.Publish(ctx, id, attempt.AttemptID, operationAttemptPlan{ + Intent: []byte("write"), Result: []byte("done"), + }) + require.NoError(t, err) + } + + var next atomic.Uint64 + var workers sync.WaitGroup + errorsCh := make(chan error, operations) + for range 8 { + workers.Add(1) + go func() { + defer workers.Done() + for { + sequence := next.Add(1) + if sequence > operations { + return + } + id := OperationID{Filesystem: first.Filesystem, Session: first.Session, Sequence: sequence} + if err := journal.Acknowledge(ctx, id); err != nil { + errorsCh <- err + } + } + }() + } + workers.Wait() + close(errorsCh) + for err := range errorsCh { + require.NoError(t, err) + } + + ts, err := journal.backend.Snapshot(ctx) + require.NoError(t, err) + session, found, err := journal.loadSessionAt(ctx, OperationSessionID{ + Filesystem: first.Filesystem, Session: first.Session, + }, ts) + require.NoError(t, err) + require.True(t, found) + require.Equal(t, operations, session.RejectBefore) + require.Empty(t, session.AckGaps) + require.Zero(t, session.Pending) + require.Zero(t, session.RetainedCount) + require.Zero(t, session.RetainedBytes) +} + func TestOperationJournalConcurrentAcknowledgementsPreserveAccounting(t *testing.T) { t.Run("memory", func(t *testing.T) { journal, _, _ := testOperationJournal(t) From 65e7a0b9dc30a316af18d0691ef26f2b443830c8 Mon Sep 17 00:00:00 2001 From: bootjp Date: Tue, 27 Oct 2026 15:02:09 +0900 Subject: [PATCH 10/11] filesystem: identify invalid receipt state Signed-off-by: bootjp --- internal/filesystem/operation_receipt.go | 4 +++- internal/filesystem/operation_record.go | 6 ++++-- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/internal/filesystem/operation_receipt.go b/internal/filesystem/operation_receipt.go index 6afbfb028..3fcc3175c 100644 --- a/internal/filesystem/operation_receipt.go +++ b/internal/filesystem/operation_receipt.go @@ -18,7 +18,9 @@ func (s *Service) VerifyOperationReceipt(ctx context.Context, id OperationID) er return err } if record.State != OperationCommitted { - return errors.WithStack(ErrOperationRecordCorrupt) + return errors.Wrapf(ErrOperationRecordCorrupt, + "operation receipt is not committed: sequence=%d state=%d generation=%d", + record.ID.Sequence, record.State, record.Generation) } return nil } diff --git a/internal/filesystem/operation_record.go b/internal/filesystem/operation_record.go index 0f3a769c1..c18399b93 100644 --- a/internal/filesystem/operation_record.go +++ b/internal/filesystem/operation_record.go @@ -107,7 +107,8 @@ func encodeOperationRecord(record operationRecord) ([]byte, error) { func decodeOperationRecord(raw []byte) (operationRecord, error) { if !validOperationHeader(raw) { - return operationRecord{}, errors.WithStack(ErrOperationRecordCorrupt) + return operationRecord{}, errors.Wrapf(ErrOperationRecordCorrupt, + "invalid operation record header: bytes=%d", len(raw)) } var record operationRecord decoder := json.NewDecoder(bytes.NewReader(raw[operationHeaderBytes:])) @@ -116,7 +117,8 @@ func decodeOperationRecord(raw []byte) (operationRecord, error) { return operationRecord{}, errors.Wrap(ErrOperationRecordCorrupt, err.Error()) } if err := decoder.Decode(new(any)); !errors.Is(err, io.EOF) { - return operationRecord{}, errors.WithStack(ErrOperationRecordCorrupt) + return operationRecord{}, errors.Wrap(ErrOperationRecordCorrupt, + "operation record has trailing JSON data") } record.CommitTS = binary.BigEndian.Uint64(raw[operationCommitTSOffset:]) if err := validateDecodedOperationRecord(raw, record); err != nil { From 08a9618f496cbf9371aa5693a6735f0b12cccf60 Mon Sep 17 00:00:00 2001 From: bootjp Date: Tue, 27 Oct 2026 15:19:51 +0900 Subject: [PATCH 11/11] filesystem: exercise generation fence and ACK race through Raft Signed-off-by: bootjp --- .../operation_journal_accounting_test.go | 139 ++++++++++++++++++ 1 file changed, 139 insertions(+) diff --git a/internal/filesystem/operation_journal_accounting_test.go b/internal/filesystem/operation_journal_accounting_test.go index daa5b0cf5..28ba02ec8 100644 --- a/internal/filesystem/operation_journal_accounting_test.go +++ b/internal/filesystem/operation_journal_accounting_test.go @@ -5,7 +5,12 @@ import ( "sync" "sync/atomic" "testing" + "time" + "github.com/bootjp/elastickv/distribution" + "github.com/bootjp/elastickv/internal/raftengine" + etcdraftengine "github.com/bootjp/elastickv/internal/raftengine/etcd" + "github.com/bootjp/elastickv/kv" "github.com/bootjp/elastickv/store" "github.com/stretchr/testify/require" ) @@ -44,6 +49,140 @@ func TestOperationJournalConcurrentAcknowledgementChurn(t *testing.T) { }) } +func TestOperationJournalGenerationFenceRacesAcknowledgement(t *testing.T) { + t.Run("memory", func(t *testing.T) { + journal, _, _ := testOperationJournal(t) + assertGenerationFenceRacesAcknowledgement(t, journal) + }) + t.Run("pebble", func(t *testing.T) { + st, err := store.NewPebbleStore(t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, st.Close()) }) + service, err := NewService(st, &testCoordinator{st: st}) + require.NoError(t, err) + journal, err := NewOperationJournal(service) + require.NoError(t, err) + assertGenerationFenceRacesAcknowledgement(t, journal) + }) + t.Run("raft-pebble", func(t *testing.T) { + journal := testRaftPebbleOperationJournal(t) + assertGenerationFenceRacesAcknowledgement(t, journal) + }) +} + +func testRaftPebbleOperationJournal(t *testing.T) *OperationJournal { + t.Helper() + st, err := store.NewPebbleStore(t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, st.Close()) }) + factory := etcdraftengine.NewFactory(etcdraftengine.FactoryConfig{ + TickInterval: 10 * time.Millisecond, HeartbeatTick: 1, ElectionTick: 10, + MaxSizePerMsg: 1 << 20, MaxInflightMsg: 256, + }) + result, err := factory.Create(raftengine.FactoryConfig{ + LocalID: "fs-ack-accounting", LocalAddress: "fs-ack-accounting", + DataDir: t.TempDir(), Bootstrap: true, + StateMachine: kv.NewKvFSMWithHLC(st, kv.NewHLC()), + }) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, result.Engine.Close()) + if result.Close != nil { + require.NoError(t, result.Close()) + } + }) + require.Eventually(t, func() bool { return result.Engine.State() == raftengine.StateLeader }, + 5*time.Second, 10*time.Millisecond) + routes := distribution.NewEngineWithDefaultRoute() + require.NoError(t, routes.ApplySnapshot(distribution.CatalogSnapshot{ + Version: 1, + Routes: []distribution.RouteDescriptor{{ + RouteID: 1, GroupID: 1, State: distribution.RouteStateActive, + }}, + })) + groups := map[uint64]*kv.ShardGroup{1: { + Engine: result.Engine, Store: st, Txn: kv.NewLeaderProxyWithEngine(result.Engine), + }} + shards := kv.NewShardStore(routes, groups) + coord := kv.NewShardedCoordinator(routes, groups, 1, kv.NewHLC(), shards) + service, err := NewService(shards, coord) + require.NoError(t, err) + require.NoError(t, service.InitializeRoot(context.Background(), testRootMode, 1000, 1000)) + journal, err := NewOperationJournal(service) + require.NoError(t, err) + return journal +} + +func assertGenerationFenceRacesAcknowledgement(t *testing.T, journal *OperationJournal) { + t.Helper() + ctx := context.Background() + pendingID := testOperationID(t, journal, 1) + terminalID := OperationID{Filesystem: pendingID.Filesystem, Session: pendingID.Session, Sequence: 2} + _, err := journal.Admit(ctx, pendingID, []byte("write"), 128, 1024) + require.NoError(t, err) + pending, err := journal.BeginAttempt(ctx, pendingID) + require.NoError(t, err) + _, err = journal.Admit(ctx, terminalID, []byte("write"), 128, 1024) + require.NoError(t, err) + terminal, err := journal.BeginAttempt(ctx, terminalID) + require.NoError(t, err) + publicationRoute, ready := journal.backend.JournalRouteVersion() + require.True(t, ready) + _, err = journal.Publish(ctx, terminalID, terminal.AttemptID, operationAttemptPlan{ + Intent: []byte("write"), Result: []byte("done"), RouteVersion: publicationRoute, + }) + require.NoError(t, err) + + sessionID := OperationSessionID{Filesystem: pendingID.Filesystem, Session: pendingID.Session} + backend := &synchronizeSessionReadsBackend{ + operationJournalBackend: journal.backend, + sessionKey: sessionID.key(), + both: make(chan struct{}), + } + concurrent := newOperationJournalWithBackend(backend) + observed, ready := backend.JournalRouteVersion() + require.True(t, ready) + type result struct { + record operationRecord + err error + } + fenceDone := make(chan result, 1) + ackDone := make(chan error, 1) + go func() { + record, fenceErr := concurrent.advanceGenerationOnce(ctx, pendingID, pending.AttemptID, observed) + fenceDone <- result{record: record, err: fenceErr} + }() + go func() { ackDone <- concurrent.acknowledgeOnce(ctx, terminalID) }() + fence := <-fenceDone + ackErr := <-ackDone + require.True(t, (fence.err == nil) != (ackErr == nil), "fence: %v; ack: %v", fence.err, ackErr) + + var fenced operationRecord + if fence.err != nil { + require.True(t, IsWriteConflict(fence.err), "fence: %v", fence.err) + fenced, err = journal.AdvanceGeneration(ctx, pendingID, pending.AttemptID) + require.NoError(t, err) + } else { + fenced = fence.record + } + if ackErr != nil { + require.True(t, IsWriteConflict(ackErr), "ack: %v", ackErr) + require.NoError(t, journal.Acknowledge(ctx, terminalID)) + } + require.Equal(t, pending.Reservation+operationFenceReservation, fenced.Reservation) + + ts, err := journal.backend.Snapshot(ctx) + require.NoError(t, err) + session, found, err := journal.loadSessionAt(ctx, sessionID, ts) + require.NoError(t, err) + require.True(t, found) + require.Equal(t, uint32(1), session.RetainedCount) + require.Equal(t, fenced.Reservation, session.RetainedBytes) + _, err = journal.Cancel(ctx, pendingID) + require.NoError(t, err) + require.NoError(t, journal.Acknowledge(ctx, pendingID)) +} + func assertConcurrentAcknowledgementChurn(t *testing.T, journal *OperationJournal) { t.Helper() ctx := context.Background()