From 66da054d1e62e26592d9c17908d95d766e8dd49a Mon Sep 17 00:00:00 2001 From: Yacov Manevich Date: Fri, 31 Jul 2026 00:08:03 +0200 Subject: [PATCH] Garbage collect the WAL periodically This commit adds periodical GC to the WAL, every WALGCInterval rounds. Signed-off-by: Yacov Manevich --- common/encoding.go | 156 ++++++++++++++----- common/encoding_maxalloc_test.go | 104 +++++++++---- common/encoding_test.go | 242 +++++++++++++++++++++++++---- config.go | 4 + instance.go | 62 ++++++-- instance_helpers_test.go | 132 +++++++++++++++- instance_test.go | 219 ++++++++++++++++++++++++++ testutil/wal.go | 5 + util.go | 10 +- util_test.go | 10 +- wal/gc.go | 86 +++++++---- wal/gc_test.go | 256 +++++++++++++++++++++++++++++-- wal/last_record.go | 4 +- 13 files changed, 1125 insertions(+), 165 deletions(-) diff --git a/common/encoding.go b/common/encoding.go index 35396800..e01fe514 100644 --- a/common/encoding.go +++ b/common/encoding.go @@ -256,7 +256,8 @@ func ParseEmptyVoteRecord(rawEmptyVote []byte) (ToBeSignedEmptyVote, error) { return emptyVote, nil } -func BlockRecordRetentionTerm(record []byte) (uint64, error) { +// blockRecordHeader extracts the block header of a block record without deserializing the block itself. +func blockRecordHeader(record []byte) (BlockHeader, error) { initialSize := len(record) var pos int @@ -266,30 +267,47 @@ func BlockRecordRetentionTerm(record []byte) (uint64, error) { pos += 4 if len(record) < pos { - return 0, fmt.Errorf("record too short to extract metadata size, expected at least %d bytes, got %d", pos, initialSize) + return BlockHeader{}, fmt.Errorf("record too short to extract metadata size, expected at least %d bytes, got %d", pos, initialSize) } metadataSize := binary.BigEndian.Uint32(record[2:6]) if metadataSize > maxAllocationSize { - return 0, fmt.Errorf("metadata size too large: %d bytes", metadataSize) + return BlockHeader{}, fmt.Errorf("metadata size too large: %d bytes", metadataSize) } record = record[6:] // Move the slice to start after the metadata size field if len(record) < int(metadataSize) { - return 0, fmt.Errorf("record too short to extract round, expected at least %d bytes, got %d", 6+int(metadataSize), initialSize) + return BlockHeader{}, fmt.Errorf("record too short to extract round, expected at least %d bytes, got %d", 6+int(metadataSize), initialSize) } var blockHeader BlockHeader if err := blockHeader.FromBytes(record[0:metadataSize]); err != nil { - return 0, fmt.Errorf("failed to deserialize block metadata: %w", err) + return BlockHeader{}, fmt.Errorf("failed to deserialize block metadata: %w", err) } + return blockHeader, nil +} + +func BlockRecordRound(record []byte) (uint64, error) { + blockHeader, err := blockRecordHeader(record) + if err != nil { + return 0, err + } return blockHeader.Round, nil } -func toBeSignedVoteQuorumRecordRetentionTerm(record []byte) (uint64, error) { +func BlockRecordEpoch(record []byte) (uint64, error) { + blockHeader, err := blockRecordHeader(record) + if err != nil { + return 0, err + } + return blockHeader.Epoch, nil +} + +// toBeSignedVoteQuorumRecordVote extracts the vote of a notarization or finalization record without deserializing its QC. +func toBeSignedVoteQuorumRecordVote(record []byte) (ToBeSignedVote, error) { initialSize := len(record) var pos int @@ -300,66 +318,120 @@ func toBeSignedVoteQuorumRecordRetentionTerm(record []byte) (uint64, error) { pos += 4 if len(record) < pos { - return 0, fmt.Errorf("record too short to extract vote size, expected at least %d bytes, got %d", pos, initialSize) + return ToBeSignedVote{}, fmt.Errorf("record too short to extract vote size, expected at least %d bytes, got %d", pos, initialSize) } voteSize := binary.BigEndian.Uint32(record[2:6]) if voteSize > maxAllocationSize { - return 0, fmt.Errorf("vote size too large: %d bytes", voteSize) + return ToBeSignedVote{}, fmt.Errorf("vote size too large: %d bytes", voteSize) } record = record[pos:] // Move the slice to start after the vote size field if len(record) < int(voteSize) { - return 0, fmt.Errorf("record too short to extract vote, expected at least %d bytes, got %d", 6+int(voteSize), initialSize) + return ToBeSignedVote{}, fmt.Errorf("record too short to extract vote, expected at least %d bytes, got %d", 6+int(voteSize), initialSize) } - record = record[:voteSize] // Trim everything but the vote, we don't need anything else to extract the round + record = record[:voteSize] // Trim everything but the vote, we don't need anything else var vote ToBeSignedVote if err := vote.FromBytes(record); err != nil { - return 0, fmt.Errorf("failed to deserialize vote: %w", err) + return ToBeSignedVote{}, fmt.Errorf("failed to deserialize vote: %w", err) } + return vote, nil +} + +func toBeSignedVoteQuorumRecordRound(record []byte) (uint64, error) { + vote, err := toBeSignedVoteQuorumRecordVote(record) + if err != nil { + return 0, err + } return vote.Round, nil } -func emptyNotarizationQuorumRecordRetentionTerm(record []byte) (uint64, error) { - if len(record) < 23 { - return 0, fmt.Errorf("record too short to extract round, expected at least 23 bytes, got %d", len(record)) +func toBeSignedVoteQuorumRecordEpoch(record []byte) (uint64, error) { + vote, err := toBeSignedVoteQuorumRecordVote(record) + if err != nil { + return 0, err + } + return vote.Epoch, nil +} + +// An empty notarization record lays out as: 2 bytes of record type, 4 bytes of vote size, +// then the empty vote itself: 1 byte of version, 8 bytes of epoch and 8 bytes of round. +const ( + emptyNotarizationRecordEpochOffset = 2 + 4 + 1 + emptyNotarizationRecordRoundOffset = emptyNotarizationRecordEpochOffset + 8 + emptyNotarizationRecordMinLen = emptyNotarizationRecordRoundOffset + 8 +) + +func emptyNotarizationQuorumRecordRound(record []byte) (uint64, error) { + if len(record) < emptyNotarizationRecordMinLen { + return 0, fmt.Errorf("record too short to extract round, expected at least %d bytes, got %d", emptyNotarizationRecordMinLen, len(record)) } - var pos int - // First 2 bytes are for record type - pos += 2 - // Next 4 bytes are for the size of the vote - pos += 4 - // The next 9 bytes are for version and epoch. - pos += 9 - // The next 8 bytes are for round. - round := binary.BigEndian.Uint64(record[pos : pos+8]) - return round, nil + return binary.BigEndian.Uint64(record[emptyNotarizationRecordRoundOffset : emptyNotarizationRecordRoundOffset+8]), nil } -func EmptyVoteRecordRetentionTerm(record []byte) (uint64, error) { - if len(record) < 19 { - return 0, fmt.Errorf("record too short to extract round, expected at least 23 bytes, got %d", len(record)) +func emptyNotarizationQuorumRecordEpoch(record []byte) (uint64, error) { + if len(record) < emptyNotarizationRecordMinLen { + return 0, fmt.Errorf("record too short to extract epoch, expected at least %d bytes, got %d", emptyNotarizationRecordMinLen, len(record)) } - var pos int - // First 2 bytes are for record type - pos += 2 - // The next 9 bytes are for version and epoch. - pos += 9 - // The next 8 bytes are for round. - round := binary.BigEndian.Uint64(record[pos : pos+8]) - return round, nil + return binary.BigEndian.Uint64(record[emptyNotarizationRecordEpochOffset : emptyNotarizationRecordEpochOffset+8]), nil } -type WALRetentionReader struct{} +// An empty vote record lays out as: 2 bytes of record type, then the empty vote itself: +// 1 byte of version, 8 bytes of epoch and 8 bytes of round. +const ( + emptyVoteRecordEpochOffset = 2 + 1 + emptyVoteRecordRoundOffset = emptyVoteRecordEpochOffset + 8 + emptyVoteRecordMinLen = emptyVoteRecordRoundOffset + 8 +) + +func EmptyVoteRecordRound(record []byte) (uint64, error) { + if len(record) < emptyVoteRecordMinLen { + return 0, fmt.Errorf("record too short to extract round, expected at least %d bytes, got %d", emptyVoteRecordMinLen, len(record)) + } + + return binary.BigEndian.Uint64(record[emptyVoteRecordRoundOffset : emptyVoteRecordRoundOffset+8]), nil +} + +func EmptyVoteRecordEpoch(record []byte) (uint64, error) { + if len(record) < emptyVoteRecordMinLen { + return 0, fmt.Errorf("record too short to extract epoch, expected at least %d bytes, got %d", emptyVoteRecordMinLen, len(record)) + } + + return binary.BigEndian.Uint64(record[emptyVoteRecordEpochOffset : emptyVoteRecordEpochOffset+8]), nil +} + +type WALReader struct{} + +// Round returns the round of the given WAL entry without deserializing it in full. +func (wrr *WALReader) Round(entry []byte) (uint64, error) { + if len(entry) < 2 { + return 0, fmt.Errorf("entry too short to extract record type, expected at least 2 bytes, got %d", len(entry)) + } + + recordType := binary.BigEndian.Uint16(entry[:2]) + switch recordType { + case BlockRecordType: + return BlockRecordRound(entry) + case NotarizationRecordType, FinalizationRecordType: + return toBeSignedVoteQuorumRecordRound(entry) + case EmptyNotarizationRecordType: + return emptyNotarizationQuorumRecordRound(entry) + case EmptyVoteRecordType: + return EmptyVoteRecordRound(entry) + default: + return 0, fmt.Errorf("unknown record type %d for round extraction", recordType) + } +} -func (wrr *WALRetentionReader) RetentionTerm(entry []byte) (uint64, error) { +// Epoch returns the epoch of the given WAL entry without deserializing it in full. +func (wrr *WALReader) Epoch(entry []byte) (uint64, error) { if len(entry) < 2 { return 0, fmt.Errorf("entry too short to extract record type, expected at least 2 bytes, got %d", len(entry)) } @@ -367,14 +439,14 @@ func (wrr *WALRetentionReader) RetentionTerm(entry []byte) (uint64, error) { recordType := binary.BigEndian.Uint16(entry[:2]) switch recordType { case BlockRecordType: - return BlockRecordRetentionTerm(entry) + return BlockRecordEpoch(entry) case NotarizationRecordType, FinalizationRecordType: - return toBeSignedVoteQuorumRecordRetentionTerm(entry) + return toBeSignedVoteQuorumRecordEpoch(entry) case EmptyNotarizationRecordType: - return emptyNotarizationQuorumRecordRetentionTerm(entry) + return emptyNotarizationQuorumRecordEpoch(entry) case EmptyVoteRecordType: - return EmptyVoteRecordRetentionTerm(entry) + return EmptyVoteRecordEpoch(entry) default: - return 0, fmt.Errorf("unknown record type %d for retention term extraction", recordType) + return 0, fmt.Errorf("unknown record type %d for epoch extraction", recordType) } } diff --git a/common/encoding_maxalloc_test.go b/common/encoding_maxalloc_test.go index d511acb5..b3e54147 100644 --- a/common/encoding_maxalloc_test.go +++ b/common/encoding_maxalloc_test.go @@ -78,61 +78,105 @@ func TestParseBlockRecordErrors(t *testing.T) { } } -func TestBlockRecordRetentionTermErrors(t *testing.T) { +func TestBlockRecordRoundAndEpochErrors(t *testing.T) { bh := BlockHeader{ProtocolMetadata: ProtocolMetadata{Version: 1, Round: 42, Seq: 2, Epoch: 3}} valid, err := BlockRecord(bh, []byte{9}) require.NoError(t, err) for _, tc := range []struct { - name string - input []byte - expectedTerm uint64 - errContains string + name string + input []byte + expectedRound uint64 + expectedEpoch uint64 + errContains string }{ - {"valid returns round", valid, 42, ""}, - {"too short for metadata size", []byte{0, 0, 0}, 0, "too short to extract metadata size"}, - {"metadata size too large", sizePrefixedRecord(BlockRecordType, maxAllocationSize+1), 0, "metadata size too large"}, - {"metadata size at limit but truncated", sizePrefixedRecord(BlockRecordType, maxAllocationSize), 0, "too short to extract round"}, - {"declared size exceeds buffer", sizePrefixedRecord(BlockRecordType, 40, 0x01), 0, "too short to extract round"}, - {"invalid metadata bytes", sizePrefixedRecord(BlockRecordType, 1, 0x80), 0, "failed to deserialize block metadata"}, + {"valid returns round and epoch", valid, 42, 3, ""}, + {"too short for metadata size", []byte{0, 0, 0}, 0, 0, "too short to extract metadata size"}, + {"metadata size too large", sizePrefixedRecord(BlockRecordType, maxAllocationSize+1), 0, 0, "metadata size too large"}, + {"metadata size at limit but truncated", sizePrefixedRecord(BlockRecordType, maxAllocationSize), 0, 0, "too short to extract round"}, + {"declared size exceeds buffer", sizePrefixedRecord(BlockRecordType, 40, 0x01), 0, 0, "too short to extract round"}, + {"invalid metadata bytes", sizePrefixedRecord(BlockRecordType, 1, 0x80), 0, 0, "failed to deserialize block metadata"}, } { t.Run(tc.name, func(t *testing.T) { - term, err := BlockRecordRetentionTerm(tc.input) + round, roundErr := BlockRecordRound(tc.input) + epoch, epochErr := BlockRecordEpoch(tc.input) if tc.errContains == "" { - require.NoError(t, err) - require.Equal(t, tc.expectedTerm, term) + require.NoError(t, roundErr) + require.NoError(t, epochErr) + require.Equal(t, tc.expectedRound, round) + require.Equal(t, tc.expectedEpoch, epoch) return } - require.ErrorContains(t, err, tc.errContains) + require.ErrorContains(t, roundErr, tc.errContains) + require.ErrorContains(t, epochErr, tc.errContains) }) } } -func TestNotarizationQuorumRecordRetentionTermErrors(t *testing.T) { +func TestNotarizationQuorumRecordRoundAndEpochErrors(t *testing.T) { vote := ToBeSignedVote{BlockHeader{ProtocolMetadata: ProtocolMetadata{Version: 1, Round: 99, Seq: 2, Epoch: 3}}} valid := NewQuorumRecord([]byte{1, 2, 3}, vote.Bytes(), NotarizationRecordType) for _, tc := range []struct { - name string - input []byte - expectedTerm uint64 - errContains string + name string + input []byte + expectedRound uint64 + expectedEpoch uint64 + errContains string }{ - {"valid returns round", valid, 99, ""}, - {"too short for vote size", []byte{0, 0, 0}, 0, "too short to extract vote size"}, - {"vote size too large", sizePrefixedRecord(NotarizationRecordType, maxAllocationSize+1), 0, "vote size too large"}, - {"vote size at limit but truncated", sizePrefixedRecord(NotarizationRecordType, maxAllocationSize), 0, "too short to extract vote,"}, - {"declared size exceeds buffer", sizePrefixedRecord(NotarizationRecordType, 40, 0x01), 0, "too short to extract vote,"}, - {"invalid vote bytes", sizePrefixedRecord(NotarizationRecordType, 1, 0x80), 0, "failed to deserialize vote"}, + {"valid returns round and epoch", valid, 99, 3, ""}, + {"too short for vote size", []byte{0, 0, 0}, 0, 0, "too short to extract vote size"}, + {"vote size too large", sizePrefixedRecord(NotarizationRecordType, maxAllocationSize+1), 0, 0, "vote size too large"}, + {"vote size at limit but truncated", sizePrefixedRecord(NotarizationRecordType, maxAllocationSize), 0, 0, "too short to extract vote,"}, + {"declared size exceeds buffer", sizePrefixedRecord(NotarizationRecordType, 40, 0x01), 0, 0, "too short to extract vote,"}, + {"invalid vote bytes", sizePrefixedRecord(NotarizationRecordType, 1, 0x80), 0, 0, "failed to deserialize vote"}, } { t.Run(tc.name, func(t *testing.T) { - term, err := toBeSignedVoteQuorumRecordRetentionTerm(tc.input) + round, roundErr := toBeSignedVoteQuorumRecordRound(tc.input) + epoch, epochErr := toBeSignedVoteQuorumRecordEpoch(tc.input) if tc.errContains == "" { - require.NoError(t, err) - require.Equal(t, tc.expectedTerm, term) + require.NoError(t, roundErr) + require.NoError(t, epochErr) + require.Equal(t, tc.expectedRound, round) + require.Equal(t, tc.expectedEpoch, epoch) return } - require.ErrorContains(t, err, tc.errContains) + require.ErrorContains(t, roundErr, tc.errContains) + require.ErrorContains(t, epochErr, tc.errContains) + }) + } +} + +func TestEmptyRecordsRoundAndEpochLengthChecks(t *testing.T) { + emptyVote := NewEmptyVoteRecord(ToBeSignedEmptyVote{EmptyVoteMetadata{Round: 7, Epoch: 5}}) + require.Len(t, emptyVote, emptyVoteRecordMinLen) + + en := &EmptyNotarization{Vote: ToBeSignedEmptyVote{EmptyVoteMetadata: EmptyVoteMetadata{Round: 8, Epoch: 6}}} + emptyNotarization := NewQuorumRecord(nil, en.Vote.Bytes(), EmptyNotarizationRecordType) + require.Len(t, emptyNotarization, emptyNotarizationRecordMinLen) + + for _, tc := range []struct { + name string + round func([]byte) (uint64, error) + epoch func([]byte) (uint64, error) + expectedRound, expectedEpoch uint64 + valid []byte + }{ + {"empty vote", EmptyVoteRecordRound, EmptyVoteRecordEpoch, 7, 5, emptyVote}, + {"empty notarization", emptyNotarizationQuorumRecordRound, emptyNotarizationQuorumRecordEpoch, 8, 6, emptyNotarization}, + } { + t.Run(tc.name, func(t *testing.T) { + round, err := tc.round(tc.valid) + require.NoError(t, err) + epoch, err := tc.epoch(tc.valid) + require.NoError(t, err) + require.NotEqual(t, round, epoch, "the round and epoch offsets must differ") + + truncated := tc.valid[:len(tc.valid)-1] + _, err = tc.round(truncated) + require.ErrorContains(t, err, "too short to extract round") + _, err = tc.epoch(truncated) + require.ErrorContains(t, err, "too short to extract epoch") }) } } diff --git a/common/encoding_test.go b/common/encoding_test.go index db1a8396..dce1cc03 100644 --- a/common/encoding_test.go +++ b/common/encoding_test.go @@ -80,9 +80,9 @@ func TestBlockRecord(t *testing.T) { record, err := BlockRecord(bh, payload) require.NoError(t, err) - retentionTerm, err := BlockRecordRetentionTerm(record) + extractedRound, err := BlockRecordRound(record) require.NoError(t, err) - require.Equal(t, uint64(666), retentionTerm) + require.Equal(t, uint64(666), extractedRound) md2, payload2, err := ParseBlockRecord(record) require.NoError(t, err) @@ -107,11 +107,20 @@ func FuzzBlockRecord(f *testing.F) { } record, err := BlockRecord(bh, payload) require.NoError(t, err) - retentionTerm, err := BlockRecordRetentionTerm(record) + extractedRound, err := BlockRecordRound(record) require.NoError(t, err) - require.Equal(t, round, retentionTerm) + require.Equal(t, round, extractedRound) + extractedEpoch, err := BlockRecordEpoch(record) + require.NoError(t, err) + require.Equal(t, epoch, extractedEpoch) md2, payload2, err := ParseBlockRecord(record) + // A record carries a block, so parsing a record built from no block data is rejected, + // even though its header is still readable above. + if len(payload) == 0 { + require.ErrorContains(t, err, "expected block data") + return + } require.NoError(t, err) require.Equal(t, bh, md2) @@ -142,9 +151,9 @@ func TestNotarizationRecord(t *testing.T) { require.NoError(t, err) record := NewQuorumRecord([]byte{1, 2, 3}, vote.Bytes(), NotarizationRecordType) - retentionTerm, err := toBeSignedVoteQuorumRecordRetentionTerm(record) + extractedRound, err := toBeSignedVoteQuorumRecordRound(record) require.NoError(t, err) - require.Equal(t, uint64(666), retentionTerm) + require.Equal(t, uint64(666), extractedRound) qc, vote2, err := ParseNotarizationRecord(record) require.NoError(t, err) @@ -171,9 +180,12 @@ func FuzzNotarizationRecord(f *testing.F) { } record := NewQuorumRecord([]byte{1, 2, 3}, vote.Bytes(), NotarizationRecordType) - retentionTerm, err := toBeSignedVoteQuorumRecordRetentionTerm(record) + extractedRound, err := toBeSignedVoteQuorumRecordRound(record) require.NoError(t, err) - require.Equal(t, round, retentionTerm) + require.Equal(t, round, extractedRound) + extractedEpoch, err := toBeSignedVoteQuorumRecordEpoch(record) + require.NoError(t, err) + require.Equal(t, epoch, extractedEpoch) qc, vote2, err := ParseNotarizationRecord(record) require.NoError(t, err) @@ -190,15 +202,65 @@ func TestEmptyVote(t *testing.T) { }, }) - retentionTerm, err := EmptyVoteRecordRetentionTerm(record) + extractedRound, err := EmptyVoteRecordRound(record) require.NoError(t, err) - require.Equal(t, uint64(666), retentionTerm) + require.Equal(t, uint64(666), extractedRound) - _, err = EmptyVoteRecordRetentionTerm(record[2:]) + extractedEpoch, err := EmptyVoteRecordEpoch(record) + require.NoError(t, err) + require.Equal(t, uint64(4), extractedEpoch) + + _, err = EmptyVoteRecordRound(record[2:]) require.ErrorContains(t, err, "record too short to extract round") + + _, err = EmptyVoteRecordEpoch(record[2:]) + require.ErrorContains(t, err, "record too short to extract epoch") +} + +func FuzzWALReaderEmptyVoteRecord(f *testing.F) { + f.Fuzz(func(t *testing.T, round, epoch uint64) { + record := NewEmptyVoteRecord(ToBeSignedEmptyVote{ + EmptyVoteMetadata{Round: round, Epoch: epoch}, + }) + + var rr WALReader + extractedRound, err := rr.Round(record) + require.NoError(t, err) + require.Equal(t, round, extractedRound) + + extractedEpoch, err := rr.Epoch(record) + require.NoError(t, err) + require.Equal(t, epoch, extractedEpoch) + + vote, err := ParseEmptyVoteRecord(record) + require.NoError(t, err) + require.Equal(t, EmptyVoteMetadata{Round: round, Epoch: epoch}, vote.EmptyVoteMetadata) + }) +} + +func FuzzWALReaderEmptyNotarizationRecord(f *testing.F) { + f.Fuzz(func(t *testing.T, round, epoch uint64, qc []byte) { + en := &EmptyNotarization{ + Vote: ToBeSignedEmptyVote{EmptyVoteMetadata: EmptyVoteMetadata{Round: round, Epoch: epoch}}, + } + record := NewQuorumRecord(qc, en.Vote.Bytes(), EmptyNotarizationRecordType) + + var rr WALReader + extractedRound, err := rr.Round(record) + require.NoError(t, err) + require.Equal(t, round, extractedRound) + + extractedEpoch, err := rr.Epoch(record) + require.NoError(t, err) + require.Equal(t, epoch, extractedEpoch) + + _, vote, err := ParseEmptyNotarizationRecord(record) + require.NoError(t, err) + require.Equal(t, EmptyVoteMetadata{Round: round, Epoch: epoch}, vote.EmptyVoteMetadata) + }) } -func TestRetentionTerm(t *testing.T) { +func TestRound(t *testing.T) { emptyVoteRecord := NewEmptyVoteRecord(ToBeSignedEmptyVote{ EmptyVoteMetadata{ Round: 669, @@ -252,7 +314,115 @@ func TestRetentionTerm(t *testing.T) { for _, tc := range []struct { name string record []byte - expectedTerm uint64 + expectedRound uint64 + expectedError string + }{ + { + name: "not enough bytes", + record: []byte{1}, + expectedError: "entry too short to extract record", + }, + { + name: "wrong record type", + record: unknownRecord, + expectedError: "unknown record type", + }, + { + name: "FinalizationRecord", + record: finalizationRecord, + expectedRound: 670, + }, + { + name: "EmptyVoteRecord", + record: emptyVoteRecord, + expectedRound: 669, + }, + { + name: "NotarizationRecord", + record: notarizationRecord, + expectedRound: 666, + }, + { + name: "EmptyNotarizationRecord", + record: emptyNotarizationRecord, + expectedRound: 667, + }, + { + name: "BlockRecord", + record: blockRecord, + expectedRound: 668, + }, + } { + t.Run(tc.name, func(t *testing.T) { + var rr WALReader + extractedRound, err := rr.Round(tc.record) + if tc.expectedError == "" { + require.NoError(t, err) + require.Equal(t, tc.expectedRound, extractedRound) + } else { + require.ErrorContains(t, err, tc.expectedError) + } + }) + } +} + +func TestEpoch(t *testing.T) { + // Every record carries a distinct epoch, so that a record read at a wrong offset is caught. + emptyVoteRecord := NewEmptyVoteRecord(ToBeSignedEmptyVote{ + EmptyVoteMetadata{ + Round: 669, + Epoch: 5, + }, + }) + + vote := ToBeSignedVote{ + BlockHeader{ + ProtocolMetadata: ProtocolMetadata{ + Version: 1, + Round: 666, + Seq: 3, + Epoch: 6, + }, + }, + } + + en := &EmptyNotarization{ + Vote: ToBeSignedEmptyVote{EmptyVoteMetadata: EmptyVoteMetadata{ + Round: 667, + Epoch: 7, + }}, + } + emptyNotarizationRecord := NewQuorumRecord([]byte{1, 2, 3}, en.Vote.Bytes(), EmptyNotarizationRecordType) + notarizationRecord := NewQuorumRecord([]byte{1, 2, 3}, vote.Bytes(), NotarizationRecordType) + blockRecord, err := BlockRecord(BlockHeader{ + ProtocolMetadata: ProtocolMetadata{ + Version: 1, + Round: 668, + Seq: 3, + Epoch: 8, + }, + }, []byte{11, 12, 13, 14, 15, 16}) + require.NoError(t, err) + + finalization := ToBeSignedFinalization{ + BlockHeader{ + ProtocolMetadata: ProtocolMetadata{ + Version: 1, + Round: 670, + Seq: 3, + Epoch: 9, + }, + }, + } + finalizationRecord := NewQuorumRecord([]byte{1, 2, 3}, finalization.Bytes(), FinalizationRecordType) + + // A record whose type is not any of the known WAL record types. + unknownRecord := NewQuorumRecord([]byte{1, 2, 3}, en.Vote.Bytes(), 9999) + + for _, tc := range []struct { + name string + record []byte + expectedEpoch uint64 expectedError string }{ { @@ -266,37 +436,47 @@ func TestRetentionTerm(t *testing.T) { expectedError: "unknown record type", }, { - name: "FinalizationRecord", - record: finalizationRecord, - expectedTerm: 670, + name: "truncated EmptyVoteRecord", + record: emptyVoteRecord[:len(emptyVoteRecord)-1], + expectedError: "record too short to extract epoch", + }, + { + name: "truncated EmptyNotarizationRecord", + record: emptyNotarizationRecord[:emptyNotarizationRecordMinLen-1], + expectedError: "record too short to extract epoch", + }, + { + name: "FinalizationRecord", + record: finalizationRecord, + expectedEpoch: 9, }, { - name: "EmptyVoteRecord", - record: emptyVoteRecord, - expectedTerm: 669, + name: "EmptyVoteRecord", + record: emptyVoteRecord, + expectedEpoch: 5, }, { - name: "NotarizationRecord", - record: notarizationRecord, - expectedTerm: 666, + name: "NotarizationRecord", + record: notarizationRecord, + expectedEpoch: 6, }, { - name: "EmptyNotarizationRecord", - record: emptyNotarizationRecord, - expectedTerm: 667, + name: "EmptyNotarizationRecord", + record: emptyNotarizationRecord, + expectedEpoch: 7, }, { - name: "BlockRecord", - record: blockRecord, - expectedTerm: 668, + name: "BlockRecord", + record: blockRecord, + expectedEpoch: 8, }, } { t.Run(tc.name, func(t *testing.T) { - var rr WALRetentionReader - retentionTerm, err := rr.RetentionTerm(tc.record) + var rr WALReader + extractedEpoch, err := rr.Epoch(tc.record) if tc.expectedError == "" { require.NoError(t, err) - require.Equal(t, tc.expectedTerm, retentionTerm) + require.Equal(t, tc.expectedEpoch, extractedEpoch) } else { require.ErrorContains(t, err, tc.expectedError) } diff --git a/config.go b/config.go index da0274f6..fb2b5c6b 100644 --- a/config.go +++ b/config.go @@ -20,6 +20,10 @@ type ParameterConfig struct { MaxNetworkDelay time.Duration // MaxRoundWindow is the maximum number of rounds that can be stored in memory. MaxRoundWindow uint64 + // WALGCInterval is the interval at which the write-ahead log is garbage collected. + // The WAL is garbage collected upon every WALGCInterval blocks. + // If WALGCInterval is 0, the WAL is never garbage collected. + WALGCInterval int } // PlatformChain is an interface that abstracts the interaction with the P-chain. diff --git a/instance.go b/instance.go index bd7e76bd..9a904e14 100644 --- a/instance.go +++ b/instance.go @@ -67,7 +67,7 @@ type Instance struct { started bool cs *CachedStorage transitionListener *epochTransitionListener - wal *wal.GarbageCollectedWAL + wal wal.GarbageCollectableWAL msm *metadata.StateMachine e *simplex.Epoch nv *nonvalidator.NonValidator @@ -136,7 +136,7 @@ func (i *Instance) maybeReplicateEpochs() error { return err } - latestIndexedEpochValidators, err := getLastAcceptedValidatorSet(&i.Config) + latestIndexedEpochValidators, epochNum, err := getLastAcceptedValidatorSet(&i.Config) if err != nil { return err } @@ -147,7 +147,7 @@ func (i *Instance) maybeReplicateEpochs() error { inLatestSet := latestValidatorSet.Nodes().Contains(i.Config.ID) if upToDate && inLatestSet { i.Config.Logger.Debug("Skipping validator set replication, because the latest validator set is up to date with the Platform Chain") - return i.startValidator(latestIndexedEpochValidators) + return i.startValidator(latestIndexedEpochValidators, epochNum) } // Start as non-validator if our last indexed validator set does not equal, the latest p-chain validator set @@ -158,8 +158,8 @@ func (i *Instance) maybeReplicateEpochs() error { return i.startNonValidator(false) } -func (i *Instance) startValidator(validators common.Nodes) error { - epochConfig, err := i.createEpochConfig(validators) +func (i *Instance) startValidator(validators common.Nodes, epochNumber uint64) error { + epochConfig, err := i.createEpochConfig(validators, epochNumber) if err != nil { return err } @@ -489,10 +489,10 @@ func (i *Instance) processEpochChange(epochChange epochChange) { case runningNonValidator: // Stop the non-validator before doing anything else, so that we don't process any more messages while we are changing epochs. i.stopNonValidator() - err = i.startAtEpoch(epochChange.validators) + err = i.startAtEpoch(epochChange.validators, epochChange.epoch) case runningValidator: i.stopValidator(true) - err = i.startAtEpoch(epochChange.validators) + err = i.startAtEpoch(epochChange.validators, epochChange.epoch) default: // This should never happen, but we log it just in case. i.lock.Unlock() i.Config.Logger.Fatal("We are not running either a validator or non-validator") @@ -506,20 +506,29 @@ func (i *Instance) processEpochChange(epochChange epochChange) { } } -func (i *Instance) createEpochConfig(validators common.Nodes) (*epochConfig, error) { - wal, err := wal.NewGarbageCollectedWAL(i.Config.WALs, i.Config.WalCreator, &common.WALRetentionReader{}, i.Config.ParameterConfig.WALMaxSizeBytes) +func (i *Instance) createEpochConfig(validators common.Nodes, epochNumber uint64) (*epochConfig, error) { + reader := &common.WALReader{} + + gcWAL, err := wal.NewGarbageCollectedWAL(i.Config.WALs, i.Config.WalCreator, reader, i.Config.ParameterConfig.WALMaxSizeBytes) if err != nil { return nil, fmt.Errorf("error creating garbage collected wal: %w", err) } - i.wal = wal + i.wal = gcWAL // We might have crashed right after a sealing block was persisted to storage, // but before the WAL was garbage collected. // In that case, we need to garbage collect the WAL to remove all entries from previous epochs. - if err := i.maybeGarbageCollectWAL(); err != nil { + if err := i.maybeGarbageCollectWALAfterSealingBlockCommit(); err != nil { return nil, err } + // Ignore records in the WAL that are of old epochs. + i.wal = &wal.EpochAwareWAL{ + GarbageCollectableWAL: gcWAL, + Reader: reader, + LowestEpoch: epochNumber, + } + numBlocks := i.Config.Storage.NumBlocks() if numBlocks == 0 { return nil, errors.New("no blocks indexed in storage") @@ -569,7 +578,11 @@ func (i *Instance) createEpochConfig(validators common.Nodes) (*epochConfig, err // set the handle approval method so that the MSM can receive self approvals i.transitionListener.handleApproval = msm.HandleApproval + garbageCollectWAL := garbageCollectWALOnIndex(i.Config.ParameterConfig.WALGCInterval, i.wal) instanceStorage := NewCallbackStorage(i.cs, msm, func(block *ParsedBlock) error { + if err := garbageCollectWAL(block); err != nil { + return err + } if err := msm.OnBlockIndex(block.StateMachineBlock); err != nil { return err } @@ -596,7 +609,7 @@ func (i *Instance) createEpochConfig(validators common.Nodes) (*epochConfig, err MaxRoundWindow: i.Config.ParameterConfig.MaxRoundWindow, ID: i.Config.ID, RandomSource: source, // Seed the random source from crypto/rand - WAL: wal, + WAL: i.wal, Logger: i.Config.Logger, SignatureAggregatorCreator: i.Config.CryptoOps.CreateSignatureAggregator, QCDeserializer: i.Config.CryptoOps, @@ -613,7 +626,26 @@ func (i *Instance) createEpochConfig(validators common.Nodes) (*epochConfig, err }, nil } -func (i *Instance) maybeGarbageCollectWAL() error { +// garbageCollectWALOnIndex returns an index hook that garbage collects the WAL +// upon every walInterval rounds, discarding the WAL files whose entries all belong +// to rounds below the one of the indexed block. A non-positive walInterval disables it. +func garbageCollectWALOnIndex(walInterval int, wal wal.GarbageCollectableWAL) func(block *ParsedBlock) error { + return func(block *ParsedBlock) error { + round := block.BlockHeader().Round + if round == 0 { + return nil + } + if walInterval <= 0 { + return nil + } + if round%uint64(walInterval) == 0 { + return wal.GarbageCollect(round) + } + return nil + } +} + +func (i *Instance) maybeGarbageCollectWALAfterSealingBlockCommit() error { lastNonSimplexHeight := i.Config.LastNonSimplexInnerBlock.Height() numBlocks := i.Config.Storage.NumBlocks() @@ -642,9 +674,9 @@ func (i *Instance) maybeGarbageCollectWAL() error { } // startAtEpoch starts either a validator or non-validator at `epoch“. -func (i *Instance) startAtEpoch(validators common.Nodes) error { +func (i *Instance) startAtEpoch(validators common.Nodes, epochNumber uint64) error { if validators.Contains(i.Config.ID) { - return i.startValidator(validators) + return i.startValidator(validators, epochNumber) } return i.startNonValidator(true) diff --git a/instance_helpers_test.go b/instance_helpers_test.go index 1ad0971b..91aebc76 100644 --- a/instance_helpers_test.go +++ b/instance_helpers_test.go @@ -61,10 +61,15 @@ var ( genesisBlock = &testInnerBlock{Height_: genesisPChainHeight, TS: time.Now(), Payload: []byte("genesis")} ) +// walGCInterval is the number of rounds after which instances under test garbage +// collect their WAL. +const walGCInterval = 10 + var paramConfig = ParameterConfig{ MaxNetworkDelay: 200 * time.Millisecond, MaxRoundWindow: 100, WALMaxSizeBytes: 1024, + WALGCInterval: walGCInterval, } type testPlatformChain struct { @@ -274,6 +279,9 @@ type walCreator struct { lock sync.Mutex wals []*testutil.TestWAL + // deleted holds, in deletion order, the records of every WAL the instance deleted, + // so a test can see what the WAL held right before it was garbage collected. + deleted [][]byte } func (w *walCreator) createWAL() (wal.DeletableWAL, error) { @@ -281,7 +289,62 @@ func (w *walCreator) createWAL() (wal.DeletableWAL, error) { w.lock.Lock() w.wals = append(w.wals, tw) w.lock.Unlock() - return tw, nil + return &deleteRecordingWAL{TestWAL: tw, creator: w}, nil +} + +// deleteRecordingWAL saves the records of the WAL into its creator right before deleting them. +type deleteRecordingWAL struct { + *testutil.TestWAL + creator *walCreator +} + +func (d *deleteRecordingWAL) Delete() error { + records, err := d.ReadAll() + require.NoError(d.creator.t, err) + + d.creator.lock.Lock() + d.creator.deleted = append(d.creator.deleted, records...) + d.creator.lock.Unlock() + + return d.TestWAL.Delete() +} + +// deletedRecords returns the records of every WAL the instance deleted so far, in deletion order. +func (w *walCreator) deletedRecords() [][]byte { + w.lock.Lock() + defer w.lock.Unlock() + return append([][]byte(nil), w.deleted...) +} + +// blocksInWAL parses every block record the live WALs of this creator hold, in append order. +func (w *walCreator) blocksInWAL() []metadata.StateMachineBlock { + w.lock.Lock() + defer w.lock.Unlock() + + var records [][]byte + for _, tw := range w.wals { + walRecords, err := tw.ReadAll() + require.NoError(w.t, err) + records = append(records, walRecords...) + } + return parseBlockRecords(w.t, records) +} + +// parseBlockRecords returns the metadata of the blocks the block records among the given WAL +// records carry, in order. Inner blocks are left out, as only the metadata tells a block's type. +func parseBlockRecords(t *testing.T, records [][]byte) []metadata.StateMachineBlock { + var blocks []metadata.StateMachineBlock + for _, record := range records { + if binary.BigEndian.Uint16(record[:2]) != common.BlockRecordType { + continue + } + _, blockBytes, err := common.ParseBlockRecord(record) + require.NoError(t, err) + var rawBlock metadata.RawBlock + require.NoError(t, rawBlock.UnmarshalCanoto(blockBytes)) + blocks = append(blocks, metadata.StateMachineBlock{Metadata: rawBlock.Metadata}) + } + return blocks } // containsNotarization reports whether any WAL this creator handed out holds a @@ -385,8 +448,12 @@ func (i *instanceComm) stop() { i.wg.Wait() } -// enqueue adds a message to be sent +// enqueue adds a message to be sent, unless the network is holding messages of its kind back. func (i *instanceComm) enqueue(m inflightMessage) { + if i.n.maybeHold(i, m) { + return + } + select { case i.queue <- m: default: @@ -656,6 +723,17 @@ type network struct { nodes []node // offline nodes stay in the network but neither send nor receive messages. offline map[string]struct{} + + // hold, when set, picks the messages the network holds back instead of delivering; + // held keeps them, in send order, until releaseHeldMessages delivers them. + hold func(from common.NodeID, msg *common.Message) bool + held []heldMessage +} + +// heldMessage is a message the network held back, along with the comm that was sending it. +type heldMessage struct { + comm *instanceComm + msg inflightMessage } func (n *network) nodesSnapshot() []node { @@ -664,6 +742,48 @@ func (n *network) nodesSnapshot() []node { return append([]node(nil), n.nodes...) } +// holdMessages makes the network hold back every message the predicate selects, until +// releaseHeldMessages is called. The predicate runs on the sender's goroutine. +func (n *network) holdMessages(hold func(from common.NodeID, msg *common.Message) bool) { + n.lock.Lock() + defer n.lock.Unlock() + n.hold = hold +} + +// maybeHold holds the message back if the network is holding messages of its kind, and reports whether it did. +// The predicate runs without the network lock, so it may inspect the nodes. +func (n *network) maybeHold(comm *instanceComm, m inflightMessage) bool { + n.lock.Lock() + hold := n.hold + n.lock.Unlock() + + if hold == nil || !hold(m.from, m.msg) { + return false + } + + n.lock.Lock() + defer n.lock.Unlock() + if n.hold == nil { + // Released in the meantime: deliver rather than strand the message. + return false + } + n.held = append(n.held, heldMessage{comm: comm, msg: m}) + return true +} + +// releaseHeldMessages stops holding messages back and delivers the ones held so far, in send order. +func (n *network) releaseHeldMessages() { + n.lock.Lock() + held := n.held + n.held = nil + n.hold = nil + n.lock.Unlock() + + for _, h := range held { + h.comm.enqueue(h.msg) + } +} + func (n *network) setOffline(id common.NodeID) { n.lock.Lock() defer n.lock.Unlock() @@ -710,6 +830,8 @@ type nodeConfig struct { lastNonSimplexBlock avalanchego.VMBlock // existingNode indicates whether the node is being added to the network for the first time (false) or is a restart of an existing node (true). existingNode bool + // params overrides the shared paramConfig the instance is configured with, when non-nil. + params *ParameterConfig } // addNode creates and starts a node in the network. @@ -732,10 +854,14 @@ func (n *network) addNodeWithConfig(id common.NodeID, cfg nodeConfig) *node { if cfg.lastNonSimplexBlock != nil { lastNonSimplex = cfg.lastNonSimplexBlock } + params := paramConfig + if cfg.params != nil { + params = *cfg.params + } instance := NewInstance(Config{ LastNonSimplexInnerBlock: lastNonSimplex, - ParameterConfig: paramConfig, + ParameterConfig: params, PlatformChain: n.pChain, Broadcaster: comm, Sender: comm, diff --git a/instance_test.go b/instance_test.go index 14be8635..42c0715d 100644 --- a/instance_test.go +++ b/instance_test.go @@ -12,6 +12,7 @@ import ( metadata "github.com/ava-labs/simplex/msm" "github.com/ava-labs/simplex/simplex" "github.com/ava-labs/simplex/testutil" + "github.com/ava-labs/simplex/wal" "github.com/stretchr/testify/require" ) @@ -426,6 +427,224 @@ func TestInstanceRestartsAfterZeroBlock(t *testing.T) { network.acceptNewBlock() } +// TestInstanceGarbageCollectsWAL asserts a validator garbage collects its WAL every +// walGCInterval rounds, so the WAL never holds an entry from a round below the last +// round it garbage collected at. +func TestInstanceGarbageCollectsWAL(t *testing.T) { + const targetBlocks = uint64(100) + + validator := newNodeMapping(1) + // The validator set never changes, so the node commits every block on its own + // and the chain stays in a single Simplex epoch throughout. + pChain := newTestPChain([]metadata.NodeBLSMapping{validator}) + network := newNetwork(t, pChain) + + // A tiny WAL size makes the WAL rotate upon every entry, so garbage collection + // discards entries round by round rather than in large batches. + params := paramConfig + params.WALMaxSizeBytes = 10 + node := network.addNodeWithConfig(validator.NodeID[:], nodeConfig{params: ¶ms}).sync() + + // collectRoundsFromWAL returns the rounds of the entries the node's WAL currently + // holds, across all the WAL files it rotated through. Garbage collected files are + // deleted, hence contribute no entries. + collectRoundsFromWAL := func() []uint64 { + var reader common.WALReader + var rounds []uint64 + + node.wals.lock.Lock() + defer node.wals.lock.Unlock() + for _, w := range node.wals.wals { + entries, err := w.ReadAll() + require.NoError(t, err) + for _, entry := range entries { + round, err := reader.Round(entry) + require.NoError(t, err) + rounds = append(rounds, round) + } + } + return rounds + } + + var garbageCollectedRoundThreshold uint64 + for range targetBlocks { + block, _ := network.acceptNewBlock() + round := block.BlockHeader().Round + + // When we commit a sealing block we wipe out the WAL entirely, so ignore this corner case. + // We only care about the ordinary blocks because we're testing regular garbage collection of the WAL. + // We also ignore the first walGCInterval rounds because the WAL is not garbage collected until after that. + if block.SealingBlockInfo() != nil || round < walGCInterval { + continue + } + + // A block is indexed, which is what the commit reports, before the garbage + // collection it triggers runs. The garbage collection of the previous interval + // has certainly completed by now though, so verify the WAL against it. + roundsFromWAL := collectRoundsFromWAL() + require.NotEmpty(t, roundsFromWAL) // Just a check that the below for loop doesn't iterate over an empty slice. + for _, r := range roundsFromWAL { + require.GreaterOrEqual(t, r, garbageCollectedRoundThreshold, + "WAL holds an entry of round %d although it was garbage collected at round %d", r, garbageCollectedRoundThreshold) + } + + // We garbage collect the WAL every walGCInterval rounds, + // but we want to check the GC one round after the GC because we don't want the WAL to be empty when we check it. + // The WAL purges all entries of rounds below the round before this round. + if round > walGCInterval && round%walGCInterval == 1 { + garbageCollectedRoundThreshold = round - 1 + } + } + + // Committing targetBlocks blocks takes at least as many rounds, so the chain ran + // long enough for the WAL to have been garbage collected several times over. + require.GreaterOrEqual(t, garbageCollectedRoundThreshold, targetBlocks-2*walGCInterval) + + // The garbage collection kept the entries of the rounds at and above the round it garbage collected at. + require.NotEmpty(t, collectRoundsFromWAL()) +} + +// TestInstanceRestartSkipsSealedEpochWALRecords emulates a crash right after a sealing block +// was persisted, but before the WAL was wiped for the new epoch. The WAL then still holds the +// sealed epoch's records: the sealing block's own, and the Telocks that extended the epoch +// until the sealing block finalized. Telock rounds collide with the new epoch's first rounds, +// so replaying them would drag the node back into the sealed epoch. The restarted node must +// ignore them, start in the new epoch, and keep committing blocks. +// +// The WAL is the node's own: the test delays the finalization of the sealing block so that the +// validators extend the sealed epoch with Telocks, and then takes the WAL as the node wiped it +// when it moved to the new epoch, which is what a crash in that window would have left behind. +func TestInstanceRestartSkipsSealedEpochWALRecords(t *testing.T) { + validator := newNodeMapping(1) + leavingValidator := newNodeMapping(2) + genesisSet := metadata.NodeBLSMappings{validator, leavingValidator} + pChain := newTestPChain(genesisSet) + network := newNetwork(t, pChain) + node := network.addNode(validator.NodeID[:]) + otherNode := network.addNode(leavingValidator.NodeID[:]).sync() + network.acceptNewBlock() + + // Hold back the finalize votes for the sealing block and for whatever follows it in the + // sealed epoch. The sealing block is then notarized but not finalized, so the leaders + // must extend the epoch with Telocks. The sender has written the block to its WAL by the + // time it votes to finalize it, so the block's type is read from there. + walsOf := map[string]*walCreator{ + string(node.id): node.wals, + string(otherNode.id): otherNode.wals, + } + extendsSealedEpoch := func(from common.NodeID, round uint64) bool { + for _, block := range walsOf[string(from)].blocksInWAL() { + if block.Metadata.SimplexProtocolMetadata.Round != round { + continue + } + blockType := block.Type() + return blockType == metadata.BlockTypeSealing || blockType == metadata.BlockTypeTelock + } + return false + } + network.holdMessages(func(from common.NodeID, msg *common.Message) bool { + switch { + case msg.FinalizeVote != nil: + return extendsSealedEpoch(from, msg.FinalizeVote.Finalization.Round) + case msg.Finalization != nil: + return extendsSealedEpoch(from, msg.Finalization.Finalization.Round) + default: + return false + } + }) + + // Seal the first epoch by shrinking the validator set to our node alone, which then + // carries the new epoch on its own. + newValidatorSet := metadata.NodeBLSMappings{validator} + pChain.setValidatorSetAt(10, newValidatorSet) + pChain.advanceHeight(10) + + // Wait until our node holds a sealing block and a Telock built on top of it. + telocksInWAL := func(blocks []metadata.StateMachineBlock) (sealingRound uint64, telocks int) { + for _, block := range blocks { + switch block.Type() { + case metadata.BlockTypeSealing: + sealingRound = block.Metadata.SimplexProtocolMetadata.Round + case metadata.BlockTypeTelock: + telocks++ + } + } + return sealingRound, telocks + } + require.Eventually(t, func() bool { + _, telocks := telocksInWAL(node.wals.blocksInWAL()) + return telocks > 0 + }, 30*time.Second, 10*time.Millisecond, "the sealed epoch was never extended by a Telock") + + // Let the finalize votes through. The sealing block finalizes, our node commits it, moves to + // the new epoch and wipes its WAL, which the WAL creator records for us. + network.releaseHeldMessages() + sealingBlock := network.waitUntilSealingBlock(newValidatorSet.Nodes()) + sealingHeader := sealingBlock.BlockHeader() + sealedEpoch, newEpoch := sealingHeader.Epoch, sealingHeader.Seq + require.NotEqual(t, sealedEpoch, newEpoch) + + require.Eventually(t, func() bool { + node.inst.lock.Lock() + defer node.inst.lock.Unlock() + return node.inst.e != nil && node.inst.e.Metadata().Epoch == newEpoch + }, 20*time.Second, 10*time.Millisecond, "the node never moved to the new epoch") + + // From here on only our node matters; the one that left must not interfere with its restart. + network.setOffline(leavingValidator.NodeID[:]) + node.stop() + // A crash right after the sealing block was persisted leaves it at the tip. + require.Equal(t, sealingHeader.Seq, node.storage.NumBlocks()-1) + + // The WAL the node wiped is what it would have found on restart after such a crash: + // the sealing block and the Telocks that extended its epoch, all in the sealed epoch. + staleRecords := node.wals.deletedRecords() + staleSealingRound, staleTelocks := telocksInWAL(parseBlockRecords(t, staleRecords)) + require.Equal(t, sealingHeader.Round, staleSealingRound) + require.Positive(t, staleTelocks) + var reader common.WALReader + for _, record := range staleRecords { + epoch, err := reader.Epoch(record) + require.NoError(t, err) + require.Equal(t, sealedEpoch, epoch) + } + + staleWAL := testutil.NewTestWAL(t) + for _, record := range staleRecords { + require.NoError(t, staleWAL.Append(record)) + } + + // Come back up over the same storage with the stale WAL. Starting is where the WAL is replayed. + restarted := network.addNodeWithConfig(validator.NodeID[:], nodeConfig{ + wals: []wal.DeletableWAL{staleWAL}, storage: node.storage, existingNode: true, + }) + + restarted.inst.lock.Lock() + epoch := restarted.inst.e + restarted.inst.lock.Unlock() + require.NotNil(t, epoch, "the node should have restarted as a validator") + + // The node starts in the new epoch, past the sealing round, not in the epoch the stale records belong to. + md := epoch.Metadata() + require.Equal(t, newEpoch, md.Epoch) + require.Greater(t, md.Round, sealingHeader.Round) + + // It keeps committing blocks in the new epoch, at the sequence right after the sealing block, + // which the first Telock would have occupied had it been replayed. + network.pending.addPendingBlock() + nextSeq := sealingHeader.Seq + 1 + require.Eventually(t, func() bool { + return restarted.storage.NumBlocks() > nextSeq + }, 20*time.Second, 10*time.Millisecond, "the restarted node never committed a block in the new epoch") + + block, _, err := restarted.storage.Retrieve(nextSeq) + require.NoError(t, err) + require.Equal(t, newEpoch, block.BlockHeader().Epoch) + require.Equal(t, nextSeq, block.BlockHeader().Seq) + require.Equal(t, sealingHeader.Digest, block.BlockHeader().Prev) + require.Equal(t, metadata.BlockTypeNormal, block.(*ParsedBlock).Type()) +} + func TestInstanceDoubleStartFails(t *testing.T) { validator := newNodeMapping(1) genesisSet := []metadata.NodeBLSMapping{validator} diff --git a/testutil/wal.go b/testutil/wal.go index fc2e8907..2a3265bc 100644 --- a/testutil/wal.go +++ b/testutil/wal.go @@ -42,7 +42,12 @@ func (tw *TestWAL) Clone() *TestWAL { return wal } +// Delete discards the entries of the WAL, as deleting a real WAL file would. func (tw *TestWAL) Delete() error { + tw.lock.Lock() + defer tw.lock.Unlock() + + tw.WriteAheadLog = wal.NewMemWAL(tw.t) return nil } diff --git a/util.go b/util.go index 0fbb1a6c..ba2a78c2 100644 --- a/util.go +++ b/util.go @@ -37,10 +37,10 @@ func LastBlock(storage Storage) (metadata.StateMachineBlock, uint64, error) { // contains non-Simplex blocks, the genesis validator set is returned. If the last // block is a sealing block, the set it seals in is returned. Otherwise, the set is // loaded from the sealing block of the last block's epoch. -func getLastAcceptedValidatorSet(config *Config) (common.Nodes, error) { +func getLastAcceptedValidatorSet(config *Config) (common.Nodes, uint64, error) { lastBlock, numBlocks, err := LastBlock(config.Storage) if err != nil { - return nil, fmt.Errorf("error retrieving last block: %w", err) + return nil, 0, fmt.Errorf("error retrieving last block: %w", err) } lastNonSimplexHeight := config.LastNonSimplexInnerBlock.Height() @@ -70,17 +70,17 @@ func getLastAcceptedValidatorSet(config *Config) (common.Nodes, error) { sealingBlockSeq := epochNum sealingBlock, _, err := config.Storage.GetBlock(sealingBlockSeq) if err != nil { - return nil, fmt.Errorf("error retrieving sealing block from storage: %w", err) + return nil, 0, fmt.Errorf("error retrieving sealing block from storage: %w", err) } if sealingBlock.Metadata.SimplexEpochInfo.BlockValidationDescriptor == nil { - return nil, fmt.Errorf("%w at seq %d", errNonSealingBlock, sealingBlockSeq) + return nil, 0, fmt.Errorf("%w at seq %d", errNonSealingBlock, sealingBlockSeq) } validatorSet = constructValidatorSetFromSealingBlock(sealingBlock.Metadata.SimplexEpochInfo.BlockValidationDescriptor) nodes = validatorSetToNodes(validatorSet) config.Logger.Debug("Determined epoch and validator set from sealing block in storage", zap.Uint64("epoch", epochNum), zap.Uint64("sealingBlockSeq", sealingBlockSeq)) } - return nodes, nil + return nodes, epochNum, nil } func validatorSetToNodes(validatorSet metadata.NodeBLSMappings) common.Nodes { diff --git a/util_test.go b/util_test.go index 6661a5f9..2d4ee087 100644 --- a/util_test.go +++ b/util_test.go @@ -125,12 +125,15 @@ func TestGetLastAcceptedEpochAndValidatorSet(t *testing.T) { name string blocks []metadata.StateMachineBlock expectedNodes common.Nodes + expectedEpoch uint64 expectedErr error }{ { name: "only non-Simplex blocks starts at first Simplex height with genesis set", blocks: []metadata.StateMachineBlock{nonSimplexBlock(0)}, expectedNodes: vdrSet.Nodes(), + // The first Simplex epoch is the sequence the first Simplex block will occupy. + expectedEpoch: 1, }, { name: "multiple non-Simplex blocks start at first Simplex height with genesis set", @@ -140,6 +143,7 @@ func TestGetLastAcceptedEpochAndValidatorSet(t *testing.T) { nonSimplexBlock(2), }, expectedNodes: vdrSet.Nodes(), + expectedEpoch: 3, }, { name: "sealing block at tip starts next epoch with its descriptor set", @@ -148,6 +152,8 @@ func TestGetLastAcceptedEpochAndValidatorSet(t *testing.T) { sealingBlock(1, 2, vdrSet), }, expectedNodes: vdrSet.Nodes(), + // The epoch a sealing block starts is numbered by the sealing block's sequence. + expectedEpoch: 2, }, { name: "non-sealing tip keeps its epoch, set loaded from sealing block at seq==epoch", @@ -158,6 +164,7 @@ func TestGetLastAcceptedEpochAndValidatorSet(t *testing.T) { simplexBlock(2, 3), }, expectedNodes: vdrSet.Nodes(), + expectedEpoch: 2, }, { name: "empty storage errors", @@ -180,13 +187,14 @@ func TestGetLastAcceptedEpochAndValidatorSet(t *testing.T) { storage := &stubStorage{blocks: tt.blocks} config := epochTestConfig(t, storage, vdrSet) - nodes, err := getLastAcceptedValidatorSet(config) + nodes, epoch, err := getLastAcceptedValidatorSet(config) if tt.expectedErr != nil { require.ErrorIs(t, err, tt.expectedErr) return } require.NoError(t, err) require.Equal(t, tt.expectedNodes, nodes) + require.Equal(t, tt.expectedEpoch, epoch) }) } } diff --git a/wal/gc.go b/wal/gc.go index 7f5eee24..e0abc154 100644 --- a/wal/gc.go +++ b/wal/gc.go @@ -17,17 +17,18 @@ const ( // Creator creates a DeletableWAL. Returns an error upon failure. type Creator func() (DeletableWAL, error) -// Reader reads the retention term of a WAL payload. +// Reader reads the round and the epoch of a WAL payload. // Returns an error upon failure. type Reader interface { - RetentionTerm([]byte) (uint64, error) + Round([]byte) (uint64, error) + Epoch([]byte) (uint64, error) } -// TruncateableWAL is a DeletableWAL that can be recentlyTruncated to a given retention term. -// It removes all entries below the given retention term. +// TruncateableWAL is a DeletableWAL that can be truncated to a given round. +// It removes all entries below the given round. type TruncateableWAL interface { DeletableWAL - Truncate(retentionTerm uint64) error + Truncate(round uint64) error } // DeletableWAL is a WAL that can be deleted. @@ -38,9 +39,16 @@ type DeletableWAL interface { Delete() error } +// GarbageCollectableWAL is a WriteAheadLog that can be garbage collected. +// It may remove all entries below the given round. +type GarbageCollectableWAL interface { + common.WriteAheadLog + GarbageCollect(round uint64) error +} + type garbageCollectedWAL struct { - retentionTerm uint64 - size int + round uint64 + size int DeletableWAL } @@ -53,7 +61,7 @@ type GarbageCollectedWAL struct { // NewGarbageCollectedWAL creates a new GarbageCollectedWAL. // It takes in a list of existing WALs to load, a Creator to create new WALs, -// a Reader to read the retention term of entries, and a maximum WAL size. +// a Reader to read the round of entries, and a maximum WAL size. // A new WAL is created when the given maximum WAL size of a WAL is reached. func NewGarbageCollectedWAL(WALs []DeletableWAL, creator Creator, reader Reader, maxWALSize int) (*GarbageCollectedWAL, error) { wals := make([]garbageCollectedWAL, 0, len(WALs)) @@ -83,32 +91,32 @@ func loadGarbageCollectedWAL(wal DeletableWAL, reader Reader) (garbageCollectedW return garbageCollectedWAL{}, fmt.Errorf("error reading entries from WAL: %v", err) } - var highestRetentionTerm uint64 + var highestRound uint64 var size int for _, entry := range entries { size += len(entry) - rt, err := reader.RetentionTerm(entry) + rt, err := reader.Round(entry) if err != nil { data := base64.StdEncoding.EncodeToString(entry) - return garbageCollectedWAL{}, fmt.Errorf("error reading retention term of entry %s: %v", data, err) + return garbageCollectedWAL{}, fmt.Errorf("error reading round of entry %s: %v", data, err) } - if rt > highestRetentionTerm { - highestRetentionTerm = rt + if rt > highestRound { + highestRound = rt } } return garbageCollectedWAL{ - retentionTerm: highestRetentionTerm, - DeletableWAL: wal, - size: size, + round: highestRound, + DeletableWAL: wal, + size: size, }, nil } func (gcw *GarbageCollectedWAL) Append(payload []byte) error { - rt, err := gcw.reader.RetentionTerm(payload) + rt, err := gcw.reader.Round(payload) if err != nil { data := base64.StdEncoding.EncodeToString(payload) - return fmt.Errorf("error reading retention term of entry %s: %v", data, err) + return fmt.Errorf("error reading round of entry %s: %v", data, err) } if gcw.shouldWeCreateNewWAL(payload) { @@ -123,9 +131,9 @@ func (gcw *GarbageCollectedWAL) Append(payload []byte) error { return fmt.Errorf("error creating new WAL: %v", err) } gcw.wals = append(gcw.wals, garbageCollectedWAL{ - DeletableWAL: newWAL, - retentionTerm: rt, - size: len(payload), + DeletableWAL: newWAL, + round: rt, + size: len(payload), }) } else { // We're still using the last WAL, so we need to update its size @@ -133,8 +141,8 @@ func (gcw *GarbageCollectedWAL) Append(payload []byte) error { } last := len(gcw.wals) - 1 - if rt > gcw.wals[last].retentionTerm { - gcw.wals[last].retentionTerm = rt + if rt > gcw.wals[last].round { + gcw.wals[last].round = rt } return gcw.wals[last].Append(payload) @@ -170,12 +178,12 @@ func (gcw *GarbageCollectedWAL) ReadAll() ([][]byte, error) { return allEntries, nil } -func (gcw *GarbageCollectedWAL) GarbageCollect(retentionTerm uint64) error { +func (gcw *GarbageCollectedWAL) GarbageCollect(round uint64) error { newWALs := make([]garbageCollectedWAL, 0, len(gcw.wals)) for i, wal := range gcw.wals { - if wal.retentionTerm < retentionTerm { - // This WAL is below the retention term, we can delete it + if wal.round < round { + // This WAL is below the round, we can delete it if err := wal.Delete(); err != nil { return fmt.Errorf("error deleting WAL %d: %v", i, err) } @@ -199,3 +207,29 @@ func (gcw *GarbageCollectedWAL) Close() error { gcw.wals = nil return nil } + +// EpochAwareWAL is a GarbageCollectableWAL that ignores records with an epoch lower than its designated epoch. +type EpochAwareWAL struct { + GarbageCollectableWAL + Reader Reader + LowestEpoch uint64 +} + +func (eaw *EpochAwareWAL) ReadAll() ([][]byte, error) { + records, err := eaw.GarbageCollectableWAL.ReadAll() + if err != nil { + return nil, err + } + + var filteredRecords [][]byte + for _, record := range records { + epoch, err := eaw.Reader.Epoch(record) + if err != nil { + return nil, err + } + if epoch >= eaw.LowestEpoch { + filteredRecords = append(filteredRecords, record) + } + } + return filteredRecords, nil +} diff --git a/wal/gc_test.go b/wal/gc_test.go index ecb2f1f9..b79272d5 100644 --- a/wal/gc_test.go +++ b/wal/gc_test.go @@ -6,6 +6,8 @@ package wal_test import ( "crypto/rand" "encoding/binary" + "errors" + "slices" "testing" "github.com/ava-labs/simplex/common" @@ -14,23 +16,47 @@ import ( "github.com/stretchr/testify/require" ) +// mockWALReader reads records laid out as 8 bytes of round followed by 8 bytes of epoch. +// The garbage collection tests only ever read the round, so their records carry random +// bytes past the round; makeEpochRecord builds records that carry both. type mockWALReader struct{} -func (m *mockWALReader) RetentionTerm(entry []byte) (uint64, error) { +func (m *mockWALReader) Round(entry []byte) (uint64, error) { + if len(entry) < 8 { + return 0, errRecordTooShort + } return binary.BigEndian.Uint64(entry[:8]), nil } +func (m *mockWALReader) Epoch(entry []byte) (uint64, error) { + if len(entry) < 16 { + return 0, errRecordTooShort + } + return binary.BigEndian.Uint64(entry[8:16]), nil +} + +var errRecordTooShort = errors.New("record too short") + // makeRecord builds a payload of exactly `size` bytes whose first 8 bytes -// encode the retention term (as read back by mockWALReader). -func makeRecord(t *testing.T, retentionTerm uint64, size int) []byte { +// encode the round (as read back by mockWALReader). +func makeRecord(t *testing.T, round uint64, size int) []byte { require.GreaterOrEqual(t, size, 8) buff := make([]byte, size) - binary.BigEndian.PutUint64(buff[:8], retentionTerm) + binary.BigEndian.PutUint64(buff[:8], round) _, err := rand.Read(buff[8:]) require.NoError(t, err) return buff } +// makeEpochRecord builds a 16 byte payload encoding the given round and epoch, +// as read back by mockWALReader. +func makeEpochRecord(round, epoch uint64) []byte { + buff := make([]byte, 16) + binary.BigEndian.PutUint64(buff[:8], round) + binary.BigEndian.PutUint64(buff[8:16], epoch) + return buff +} + // countingCreator returns a Creator that records how many WALs it has created // and hands back a pointer to the running count. func countingCreator(t *testing.T) (wal.Creator, *int) { @@ -165,7 +191,7 @@ func TestGarbageCollectedWALRotatesOnOversizedPayload(t *testing.T) { } func TestGarbageCollectedWALGarbageCollectsRotatedWALs(t *testing.T) { - // Verifies a WAL is removed only once its highest retention term drops below the cutoff. + // Verifies a WAL is removed only once its highest round drops below the cutoff. const ( maxWALSize = 100 recordSize = 40 @@ -177,7 +203,7 @@ func TestGarbageCollectedWALGarbageCollectsRotatedWALs(t *testing.T) { gcw, err := wal.NewGarbageCollectedWAL(nil, creator, &mockWALReader{}, maxWALSize) require.NoError(t, err) - // Retention terms 0..5, two per WAL: + // Rounds 0..5, two per WAL: // WAL0 -> {0,1} (max 1), WAL1 -> {2,3} (max 3), WAL2 -> {4,5} (max 5) records := make([][]byte, numRecords) for rt := 0; rt < numRecords; rt++ { @@ -186,7 +212,7 @@ func TestGarbageCollectedWALGarbageCollectsRotatedWALs(t *testing.T) { } require.Equal(t, 3, *newWALCount) // 6 records, two per WAL -> 3 WALs. - // Does not drop any WALs because the cutoff is below the max retention term of WAL0. + // Does not drop any WALs because the cutoff is below the max round of WAL0. require.NoError(t, gcw.GarbageCollect(1)) got, err := gcw.ReadAll() require.NoError(t, err) @@ -198,7 +224,7 @@ func TestGarbageCollectedWALGarbageCollectsRotatedWALs(t *testing.T) { require.NoError(t, err) require.Equal(t, records[2:], got) - // Does not drop WAL1 because the cutoff is equal to the max retention term of WAL1. + // Does not drop WAL1 because the cutoff is equal to the max round of WAL1. require.NoError(t, gcw.GarbageCollect(3)) got, err = gcw.ReadAll() require.NoError(t, err) @@ -245,7 +271,7 @@ func TestGarbageCollectedWALAppendOutOfOrder(t *testing.T) { } // We create a block record for 4 and a finalization record for 3 out of order. - // The WALRetentionReader must be able to read the retention term from every + // The WALReader must be able to read the round from every // record type, or GarbageCollectedWAL.Append will fail on the finalization. records := [][]byte{ blockRecord(1), notarizationRecord(1), @@ -258,7 +284,7 @@ func TestGarbageCollectedWALAppendOutOfOrder(t *testing.T) { // A small maxWALSize forces the stream across several WAL segments so ReadAll must stitch every segment back together. creator, newWALCount := countingCreator(t) walSize := 512 - gcw, err := wal.NewGarbageCollectedWAL(nil, creator, &common.WALRetentionReader{}, walSize) + gcw, err := wal.NewGarbageCollectedWAL(nil, creator, &common.WALReader{}, walSize) require.NoError(t, err) var sum int @@ -313,3 +339,213 @@ func TestGarbageCollectedWALAppendAfterRotationAndGC(t *testing.T) { require.NoError(t, err) require.Equal(t, next, got) } + +// newEpochAwareWAL wraps a fresh garbage collected WAL that hides records of epochs below lowestEpoch. +func newEpochAwareWAL(t *testing.T, lowestEpoch uint64) (*wal.EpochAwareWAL, *wal.GarbageCollectedWAL) { + creator, _ := countingCreator(t) + gcw, err := wal.NewGarbageCollectedWAL(nil, creator, &mockWALReader{}, 100) + require.NoError(t, err) + + return &wal.EpochAwareWAL{ + GarbageCollectableWAL: gcw, + Reader: &mockWALReader{}, + LowestEpoch: lowestEpoch, + }, gcw +} + +func TestEpochAwareWALFiltersOlderEpochs(t *testing.T) { + // Rounds are not unique across epochs: the records of epoch 4 at rounds 3 and 4 collide + // with the records of epoch 5 at the same rounds, so only the epoch can tell them apart. + oldEpochs := [][]byte{ + makeEpochRecord(1, 3), + makeEpochRecord(2, 3), + makeEpochRecord(3, 4), + makeEpochRecord(4, 4), + } + currentEpoch := [][]byte{ + makeEpochRecord(3, 5), + makeEpochRecord(4, 5), + } + newerEpoch := [][]byte{ + makeEpochRecord(5, 6), + } + + eaw, gcw := newEpochAwareWAL(t, 5) + for _, record := range slices.Concat(oldEpochs, currentEpoch, newerEpoch) { + require.NoError(t, eaw.Append(record)) + } + + // Reading through the wrapper hides the older epochs and preserves the append order of the rest. + got, err := eaw.ReadAll() + require.NoError(t, err) + require.Equal(t, slices.Concat(currentEpoch, newerEpoch), got) + + // The wrapper only filters what is read; the underlying WAL still holds every record. + all, err := gcw.ReadAll() + require.NoError(t, err) + require.Equal(t, slices.Concat(oldEpochs, currentEpoch, newerEpoch), all) +} + +func TestEpochAwareWALLowestEpochZeroKeepsEverything(t *testing.T) { + records := [][]byte{ + makeEpochRecord(1, 0), + makeEpochRecord(2, 1), + makeEpochRecord(3, 7), + } + + eaw, _ := newEpochAwareWAL(t, 0) + for _, record := range records { + require.NoError(t, eaw.Append(record)) + } + + got, err := eaw.ReadAll() + require.NoError(t, err) + require.Equal(t, records, got) +} + +func TestEpochAwareWALReadAllEmpty(t *testing.T) { + eaw, _ := newEpochAwareWAL(t, 5) + + got, err := eaw.ReadAll() + require.NoError(t, err) + require.Empty(t, got) + + // A WAL holding only older epochs reads back as empty too. + require.NoError(t, eaw.Append(makeEpochRecord(1, 4))) + got, err = eaw.ReadAll() + require.NoError(t, err) + require.Empty(t, got) +} + +func TestEpochAwareWALGarbageCollectsByRound(t *testing.T) { + // Garbage collection is keyed on rounds and filtering on epochs; the two compose. + eaw, _ := newEpochAwareWAL(t, 5) + + // A tiny WAL size rotates upon every record, so garbage collection can drop records one by one. + eaw.GarbageCollectableWAL = func() *wal.GarbageCollectedWAL { + creator, _ := countingCreator(t) + gcw, err := wal.NewGarbageCollectedWAL(nil, creator, &mockWALReader{}, 16) + require.NoError(t, err) + return gcw + }() + + records := [][]byte{ + makeEpochRecord(1, 4), + makeEpochRecord(2, 5), + makeEpochRecord(3, 4), + makeEpochRecord(3, 5), + makeEpochRecord(4, 5), + } + for _, record := range records { + require.NoError(t, eaw.Append(record)) + } + + // Collecting below round 3 removes the records of rounds 1 and 2, whatever their epoch. + require.NoError(t, eaw.GarbageCollect(3)) + got, err := eaw.ReadAll() + require.NoError(t, err) + require.Equal(t, [][]byte{makeEpochRecord(3, 5), makeEpochRecord(4, 5)}, got) + + // The older epoch's record of round 3 was garbage collected too, it just was never visible. + all, err := eaw.GarbageCollectableWAL.ReadAll() + require.NoError(t, err) + require.Equal(t, records[2:], all) +} + +func TestEpochAwareWALPropagatesReaderError(t *testing.T) { + eaw, _ := newEpochAwareWAL(t, 5) + + // The record is long enough to carry a round, which Append needs, but not an epoch. + require.NoError(t, eaw.Append(makeRecord(t, 1, 8))) + + _, err := eaw.ReadAll() + require.ErrorIs(t, err, errRecordTooShort) +} + +// failingWAL fails every read with the given error. +type failingWAL struct { + wal.GarbageCollectableWAL + err error +} + +func (f *failingWAL) ReadAll() ([][]byte, error) { + return nil, f.err +} + +func TestEpochAwareWALPropagatesReadAllError(t *testing.T) { + sentinel := errors.New("disk on fire") + eaw, gcw := newEpochAwareWAL(t, 5) + eaw.GarbageCollectableWAL = &failingWAL{GarbageCollectableWAL: gcw, err: sentinel} + + _, err := eaw.ReadAll() + require.ErrorIs(t, err, sentinel) +} + +func TestEpochAwareWALWithRealRecords(t *testing.T) { + // Exercises the wrapper with the real record encodings and reader, across every record type. + // Epoch 2 was sealed at round 5, but its Telocks kept it going at rounds 6 and 7 while the + // sealing block finalized. Epoch 3 starts at round 6, reusing those rounds. + const ( + sealedEpoch = 2 + newEpoch = 3 + ) + + nodes := []common.NodeID{{1}, {2}, {3}, {4}} + sigAggr := &testutil.TestSignatureAggregator{N: len(nodes)} + + block := func(epoch, round uint64) *testutil.TestBlock { + return testutil.NewTestBlock(common.ProtocolMetadata{Epoch: epoch, Round: round, Seq: round}, common.Blacklist{}) + } + blockRecord := func(epoch, round uint64) []byte { + b := block(epoch, round) + record, err := common.BlockRecord(b.BlockHeader(), b.Bytes()) + require.NoError(t, err) + return record + } + notarizationRecord := func(epoch, round uint64) []byte { + record, err := testutil.NewNotarizationRecord(testutil.MakeLogger(t), sigAggr, block(epoch, round), nodes) + require.NoError(t, err) + return record + } + finalizationRecord := func(epoch, round uint64) []byte { + _, record := testutil.NewFinalizationRecord(t, sigAggr, block(epoch, round), nodes) + return record + } + emptyNotarizationRecord := func(epoch, round uint64) []byte { + en := testutil.NewEmptyNotarization(nodes, round) + en.Vote.Epoch = epoch + return common.NewEmptyNotarizationRecord(en) + } + emptyVoteRecord := func(epoch, round uint64) []byte { + return common.NewEmptyVoteRecord(common.ToBeSignedEmptyVote{ + EmptyVoteMetadata: common.EmptyVoteMetadata{Epoch: epoch, Round: round}, + }) + } + + sealedEpochRecords := [][]byte{ + blockRecord(sealedEpoch, 5), notarizationRecord(sealedEpoch, 5), finalizationRecord(sealedEpoch, 5), + blockRecord(sealedEpoch, 6), notarizationRecord(sealedEpoch, 6), + emptyVoteRecord(sealedEpoch, 7), emptyNotarizationRecord(sealedEpoch, 7), + } + newEpochRecords := [][]byte{ + emptyVoteRecord(newEpoch, 6), emptyNotarizationRecord(newEpoch, 6), + blockRecord(newEpoch, 7), notarizationRecord(newEpoch, 7), finalizationRecord(newEpoch, 7), + } + + creator, _ := countingCreator(t) + gcw, err := wal.NewGarbageCollectedWAL(nil, creator, &common.WALReader{}, 512) + require.NoError(t, err) + eaw := &wal.EpochAwareWAL{ + GarbageCollectableWAL: gcw, + Reader: &common.WALReader{}, + LowestEpoch: newEpoch, + } + + for _, record := range slices.Concat(sealedEpochRecords, newEpochRecords) { + require.NoError(t, eaw.Append(record)) + } + + got, err := eaw.ReadAll() + require.NoError(t, err) + require.Equal(t, newEpochRecords, got) +} diff --git a/wal/last_record.go b/wal/last_record.go index 9cb6eb2f..a091e76e 100644 --- a/wal/last_record.go +++ b/wal/last_record.go @@ -103,7 +103,7 @@ func (ri *LastRecordStoringWAL) LastRecord() []byte { return ri.lastRecord } -func (ri *LastRecordStoringWAL) Truncate(retentionTerm uint64) error { +func (ri *LastRecordStoringWAL) Truncate(round uint64) error { file, err := os.OpenFile(ri.recordIndexFilePath, os.O_CREATE|os.O_RDWR, WalPermissions) if err != nil { return fmt.Errorf("could not open last record file %s: %w", ri.recordIndexFilePath, err) @@ -123,7 +123,7 @@ func (ri *LastRecordStoringWAL) Truncate(retentionTerm uint64) error { return fmt.Errorf("could not close last record file %s: %w", ri.recordIndexFilePath, err) } - return ri.TruncateableWAL.Truncate(retentionTerm) + return ri.TruncateableWAL.Truncate(round) } func (ri *LastRecordStoringWAL) Append(record []byte) error {