diff --git a/internal/filesystem/operation_journal.go b/internal/filesystem/operation_journal.go index 26d345766..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 } @@ -449,8 +452,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_accounting_test.go b/internal/filesystem/operation_journal_accounting_test.go new file mode 100644 index 000000000..28ba02ec8 --- /dev/null +++ b/internal/filesystem/operation_journal_accounting_test.go @@ -0,0 +1,362 @@ +package filesystem + +import ( + "context" + "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" +) + +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 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 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() + 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) + 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() + 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) +} diff --git a/internal/filesystem/operation_journal_exec.go b/internal/filesystem/operation_journal_exec.go index f9c675d90..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 } @@ -665,7 +668,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_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 5a08aa245..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,19 +117,30 @@ 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 !validDecodedOperationRecord(raw, record) { - return operationRecord{}, errors.WithStack(ErrOperationRecordCorrupt) + if err := validateDecodedOperationRecord(raw, record); err != nil { + return operationRecord{}, err } 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 validateDecodedOperationRecord(raw []byte, record operationRecord) error { + if raw[4] == operationRecordLegacyVersion && (record.RecordIsPrimary || len(record.FencedAttempts) != 0) { + return errors.Wrap(ErrOperationRecordCorrupt, "legacy operation row has primary metadata") + } + if record.State != OperationPending && record.CommitTS == 0 { + return errors.Wrap(ErrOperationRecordCorrupt, "terminal operation row has no commit timestamp") + } + if record.State != OperationState(raw[5]) { + return errors.Wrap(ErrOperationRecordCorrupt, "operation row state differs from header") + } + if err := validateOperationRecord(record); err != nil { + return errors.Wrap(err, "invalid decoded operation row") + } + return nil } func validOperationHeader(raw []byte) bool { @@ -139,8 +151,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 } 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 }