Skip to content

Garbage collect the WAL periodically #476

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
yacovm wants to merge 2 commits into from
Closed

Garbage collect the WAL periodically #476

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
66da054
Garbage collect the WAL periodically
yacovm Jul 30, 2026
d5ad046
Merge branch 'main' into GC
samliok Oct 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
156 changes: 114 additions & 42 deletions common/encoding.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand All @@ -300,81 +318,135 @@ 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))
}

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)
}
}
104 changes: 74 additions & 30 deletions common/encoding_maxalloc_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
})
}
}
Loading
Loading