[Storehouse] 008 Replay WAL onto Payloadless trie

cmd/util/common/checkpoint_test.go

@@ -0,0 +1,88 @@
+package common
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/onflow/flow-go/ledger/complete/wal"
+	modelbootstrap "github.com/onflow/flow-go/model/bootstrap"
+	"github.com/onflow/flow-go/utils/unittest"
+)
+
+// TestFindLatestCheckpointFilePath verifies that the returned file name matches the version of the
+// latest checkpoint in the directory. V6 and V7 checkpoints coexist in a payloadless triedir, so
+// rendering the wrong version's name would point at a file that does not exist.
+func TestFindLatestCheckpointFilePath(t *testing.T) {
+	v6Root := modelbootstrap.FilenameWALRootCheckpoint
+	v7Root := modelbootstrap.FilenameWALRootCheckpoint + wal.V7FileSuffix
+
+	tests := []struct {
+		name     string
+		files    []string
+		expected string
+	}{
+		{
+			name:     "empty directory falls back to the V6 root checkpoint",
+			files:    nil,
+			expected: v6Root,
+		},
+		{
+			name:     "V6 root checkpoint only",
+			files:    []string{v6Root},
+			expected: v6Root,
+		},
+		{
+			name: "V7 root checkpoint is preferred over the V6 root checkpoint",
+			// this is the state of a payloadless triedir right after bootstrap: the V6 root
+			// checkpoint copied from the bootstrap folder, plus its V7 conversion
+			files:    []string{v6Root, v7Root},
+			expected: v7Root,
+		},
+		{
+			name:     "numbered V6 checkpoint",
+			files:    []string{v6Root, wal.NumberToFilename(10)},
+			expected: wal.NumberToFilename(10),
+		},
+		{
+			name:     "numbered V7 checkpoint",
+			files:    []string{v6Root, v7Root, wal.NumberToFilenameV7(10)},
+			expected: wal.NumberToFilenameV7(10),
+		},
+		{
+			name:     "highest number wins across versions",
+			files:    []string{wal.NumberToFilename(20), wal.NumberToFilenameV7(10)},
+			expected: wal.NumberToFilename(20),
+		},
+		{
+			name:     "V7 wins over V6 at the same number",
+			files:    []string{wal.NumberToFilename(10), wal.NumberToFilenameV7(10)},
+			expected: wal.NumberToFilenameV7(10),
+		},
+		{
+			name: "numbered checkpoints win over the root checkpoint",
+			files: []string{
+				v6Root, v7Root,
+				wal.NumberToFilename(10), wal.NumberToFilenameV7(10),
+				wal.NumberToFilenameV7(11),
+			},
+			expected: wal.NumberToFilenameV7(11),
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			unittest.RunWithTempDir(t, func(dir string) {
+				for _, name := range tc.files {
+					require.NoError(t, os.WriteFile(filepath.Join(dir, name), []byte{}, 0644))
+				}
+
+				checkpointFilePath, err := findLatestCheckpointFilePath(dir)
+				require.NoError(t, err)
+				require.Equal(t, filepath.Join(dir, tc.expected), checkpointFilePath)
+			})
+		})
+	}
+}

ledger/complete/ledger.go

@@ -335,15 +335,6 @@ func (l *Ledger) Trie(rootHash ledger.RootHash) (*trie.MTrie, error) {
 	return l.forest.GetTrie(rootHash)
 }
 
-// Checkpointer returns a checkpointer instance
-func (l *Ledger) Checkpointer() (*realWAL.Checkpointer, error) {
-	checkpointer, err := l.wal.NewCheckpointer()
-	if err != nil {
-		return nil, fmt.Errorf("cannot create checkpointer for compactor: %w", err)
-	}
-	return checkpointer, nil
-}
-
 func (l *Ledger) MigrateAt(
 	state ledger.State,
 	migration ledger.Migration,

ledger/complete/ledger_with_compactor.go

@@ -34,8 +34,6 @@ func NewLedgerWithCompactor(
 	logger zerolog.Logger,
 	pathFinderVersion uint8,
 ) (*LedgerWithCompactor, error) {
-	logger = logger.With().Str("ledger_mod", "complete").Logger()
-
 	// Create the ledger
 	l, err := NewLedger(diskWAL, ledgerCapacity, metrics, logger, pathFinderVersion)
 	if err != nil {

ledger/complete/payloadless/flattener_test.go

@@ -0,0 +1,139 @@
+package payloadless
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/onflow/flow-go/ledger"
+	"github.com/onflow/flow-go/ledger/common/hash"
+	"github.com/onflow/flow-go/ledger/common/testutils"
+)
+
+// noChildNodes is the getNode callback for reading leaf nodes, which never reference children.
+func noChildNodes(t *testing.T) func(nodeIndex uint64) (*Node, error) {
+	return func(nodeIndex uint64) (*Node, error) {
+		require.FailNow(t, "leaf node must not resolve child nodes", "index %d", nodeIndex)
+		return nil, nil
+	}
+}
+
+// TestEncodeDecodeLeafNodeWithLeafHash covers the `leafHashPresent` encoding: an allocated
+// register's leaf keeps its leaf hash across a round trip, and the encoding is 100 bytes.
+func TestEncodeDecodeLeafNodeWithLeafHash(t *testing.T) {
+	path := testutils.PathByUint8(7)
+	leaf := NewLeaf(path, []byte("register value"), 256)
+	require.NotNil(t, leaf.LeafHash(), "sanity check: an allocated register leaf has a leaf hash")
+
+	scratch := make([]byte, 1024)
+	encoded := EncodeNode(leaf, 0, 0, scratch)
+
+	// node type (1) + height (2) + hash (32) + path (32) + leaf hash flag (1) + leaf hash (32)
+	require.Len(t, encoded, 100)
+	require.Equal(t, leafHashPresent, encoded[encNodeTypeSize+encHeightSize+encHashSize+encPathSize])
+
+	decoded, err := ReadNode(bytes.NewReader(encoded), make([]byte, 1024), noChildNodes(t))
+	require.NoError(t, err)
+	require.Equal(t, leaf.Height(), decoded.Height())
+	require.Equal(t, leaf.Hash(), decoded.Hash())
+	require.Equal(t, *leaf.Path(), *decoded.Path())
+	require.NotNil(t, decoded.LeafHash())
+	require.Equal(t, *leaf.LeafHash(), *decoded.LeafHash())
+}
+
+// TestEncodeDecodeLeafNodeWithoutLeafHash covers the `leafHashAbsent` encoding, which is the one
+// on-disk mechanism V7 adds over V6. A leaf for an unallocated register has no leaf hash, so the
+// flag byte is the only thing recording its absence, and the encoding is 32 bytes shorter.
+func TestEncodeDecodeLeafNodeWithoutLeafHash(t *testing.T) {
+	path := testutils.PathByUint8(7)
+
+	// An unallocated register (empty value) yields a default leaf, whose leaf hash is nil.
+	leaf := NewLeaf(path, nil, 256)
+	require.True(t, leaf.IsLeaf())
+	require.Nil(t, leaf.LeafHash(), "sanity check: an unallocated register leaf has no leaf hash")
+
+	scratch := make([]byte, 1024)
+	encoded := EncodeNode(leaf, 0, 0, scratch)
+
+	// node type (1) + height (2) + hash (32) + path (32) + leaf hash flag (1), and no leaf hash
+	require.Len(t, encoded, 68)
+	require.Equal(t, leafHashAbsent, encoded[encNodeTypeSize+encHeightSize+encHashSize+encPathSize])
+
+	decoded, err := ReadNode(bytes.NewReader(encoded), make([]byte, 1024), noChildNodes(t))
+	require.NoError(t, err)
+	require.Equal(t, leaf.Height(), decoded.Height())
+	require.Equal(t, leaf.Hash(), decoded.Hash())
+	require.Equal(t, *leaf.Path(), *decoded.Path())
+	require.Nil(t, decoded.LeafHash(), "absent leaf hash must decode back to nil")
+}
+
+// TestReadNodeRejectsInvalidLeafHashFlag verifies that a leaf hash flag other than
+// `leafHashAbsent` or `leafHashPresent` is reported as an error rather than silently
+// interpreted, so a corrupted checkpoint cannot be read as a valid trie.
+func TestReadNodeRejectsInvalidLeafHashFlag(t *testing.T) {
+	leaf := NewLeaf(testutils.PathByUint8(7), []byte("register value"), 256)
+
+	encoded := EncodeNode(leaf, 0, 0, make([]byte, 1024))
+	flagPos := encNodeTypeSize + encHeightSize + encHashSize + encPathSize
+
+	for _, flag := range []byte{2, 0xff} {
+		corrupted := make([]byte, len(encoded))
+		copy(corrupted, encoded)
+		corrupted[flagPos] = flag
+
+		_, err := ReadNode(bytes.NewReader(corrupted), make([]byte, 1024), noChildNodes(t))
+		require.Error(t, err)
+		require.ErrorContains(t, err, "invalid leaf hash flag")
+	}
+}
+
+// TestReadNodeRejectsTruncatedLeafHash verifies that a leaf whose flag promises a leaf hash but
+// whose bytes are cut short is reported as an error.
+func TestReadNodeRejectsTruncatedLeafHash(t *testing.T) {
+	leaf := NewLeaf(testutils.PathByUint8(7), []byte("register value"), 256)
+
+	encoded := EncodeNode(leaf, 0, 0, make([]byte, 1024))
+	// drop the last byte of the leaf hash
+	truncated := encoded[:len(encoded)-1]
+
+	_, err := ReadNode(bytes.NewReader(truncated), make([]byte, 1024), noChildNodes(t))
+	require.Error(t, err)
+	require.ErrorContains(t, err, "cannot read leaf hash")
+}
+
+// TestEncodeDecodeLeafNodeSmallScratch verifies both leaf encodings are correct when the scratch
+// buffer is too small to hold the node, i.e. when the encoder and decoder allocate instead.
+func TestEncodeDecodeLeafNodeSmallScratch(t *testing.T) {
+	leaves := map[string]*Node{
+		"leaf hash present": NewLeaf(testutils.PathByUint8(7), []byte("register value"), 256),
+		"leaf hash absent":  NewLeaf(testutils.PathByUint8(7), nil, 256),
+	}
+
+	for name, leaf := range leaves {
+		t.Run(name, func(t *testing.T) {
+			encoded := EncodeNode(leaf, 0, 0, nil)
+
+			decoded, err := ReadNode(bytes.NewReader(encoded), nil, noChildNodes(t))
+			require.NoError(t, err)
+			require.Equal(t, leaf.Hash(), decoded.Hash())
+			require.Equal(t, leaf.LeafHash() == nil, decoded.LeafHash() == nil)
+		})
+	}
+}
+
+// TestEncodeDecodeLeafNodeWithZeroLeafHash guards against a flag-free encoding: an all-zero leaf
+// hash is a legitimate value and must not be confused with an absent one.
+func TestEncodeDecodeLeafNodeWithZeroLeafHash(t *testing.T) {
+	var zeroLeafHash hash.Hash
+	path := testutils.PathByUint8(7)
+	leaf := NewNode(256, nil, nil, path, &zeroLeafHash, ledger.GetDefaultHashForHeight(0))
+
+	encoded := EncodeNode(leaf, 0, 0, make([]byte, 1024))
+	require.Len(t, encoded, 100)
+
+	decoded, err := ReadNode(bytes.NewReader(encoded), make([]byte, 1024), noChildNodes(t))
+	require.NoError(t, err)
+	require.NotNil(t, decoded.LeafHash(), "a zero leaf hash is present, not absent")
+	require.Equal(t, zeroLeafHash, *decoded.LeafHash())
+}

ledger/complete/payloadless_compactor.go

@@ -239,58 +239,6 @@ Loop:
 	}
 }
 
-// processTrieUpdate writes the WAL record, tracks the active segment, hands
-// the newly-built trie to the queue, and signals when enough segments have
-// rolled over to checkpoint. Mirrors [Compactor.processTrieUpdate].
-func (c *PayloadlessCompactor) processTrieUpdate(
-	update *WALPayloadlessTrieUpdate,
-	trieQueue *realWAL.PayloadlessTrieQueue,
-	activeSegmentNum int,
-	nextCheckpointNum int,
-) (_activeSegmentNum int, checkpointNum int, checkpointTries []*payloadless.MTrie) {
-
-	segmentNum, skipped, updateErr := c.wal.RecordUpdate(update.Update)
-	update.ResultCh <- updateErr
-
-	defer func() {
-		// Receive the freshly-built trie from the ledger goroutine and stage it.
-		trie := <-update.TrieCh
-		if trie == nil {
-			c.logger.Error().Msg("payloadless compactor failed to get updated trie")
-			return
-		}
-		trieQueue.Push(trie)
-	}()
-
-	if activeSegmentNum == -1 {
-		return segmentNum, -1, nil
-	}
-
-	if updateErr != nil || skipped || segmentNum == activeSegmentNum {
-		return activeSegmentNum, -1, nil
-	}
-
-	// segmentNum > activeSegmentNum — a segment just rolled over.
-
-	if segmentNum != activeSegmentNum+1 {
-		c.logger.Error().Msgf("payloadless compactor got unexpected new segment %d, want %d", segmentNum, activeSegmentNum+1)
-	}
-
-	prevSegmentNum := activeSegmentNum
-	activeSegmentNum = segmentNum
-
-	c.logger.Info().Msgf("finish writing segment file %v, payloadless trie update writing to segment %v; checkpoint triggers at segment %v",
-		prevSegmentNum, activeSegmentNum, nextCheckpointNum)
-
-	if nextCheckpointNum > prevSegmentNum {
-		return activeSegmentNum, -1, nil
-	}
-
-	// nextCheckpointNum == prevSegmentNum — enough segments accumulated.
-	tries := trieQueue.Tries()
-	return activeSegmentNum, nextCheckpointNum, tries
-}
-
 // checkpoint serializes a V7 checkpoint, then prunes older V7 files per the
 // retention policy, and notifies observers.
 func (c *PayloadlessCompactor) checkpoint(ctx context.Context, tries []*payloadless.MTrie, checkpointNum int) error {
@@ -370,6 +318,58 @@ func cleanupCheckpointsV7(checkpointer *realWAL.Checkpointer, checkpointsToKeep
 	return nil
 }
 
+// processTrieUpdate writes the WAL record, tracks the active segment, hands
+// the newly-built trie to the queue, and signals when enough segments have
+// rolled over to checkpoint. Mirrors [Compactor.processTrieUpdate].
+func (c *PayloadlessCompactor) processTrieUpdate(
+	update *WALPayloadlessTrieUpdate,
+	trieQueue *realWAL.PayloadlessTrieQueue,
+	activeSegmentNum int,
+	nextCheckpointNum int,
+) (_activeSegmentNum int, checkpointNum int, checkpointTries []*payloadless.MTrie) {
+
+	segmentNum, skipped, updateErr := c.wal.RecordUpdate(update.Update)
+	update.ResultCh <- updateErr
+
+	defer func() {
+		// Receive the freshly-built trie from the ledger goroutine and stage it.
+		trie := <-update.TrieCh
+		if trie == nil {
+			c.logger.Error().Msg("payloadless compactor failed to get updated trie")
+			return
+		}
+		trieQueue.Push(trie)
+	}()
+
+	if activeSegmentNum == -1 {
+		return segmentNum, -1, nil
+	}
+
+	if updateErr != nil || skipped || segmentNum == activeSegmentNum {
+		return activeSegmentNum, -1, nil
+	}
+
+	// segmentNum > activeSegmentNum — a segment just rolled over.
+
+	if segmentNum != activeSegmentNum+1 {
+		c.logger.Error().Msgf("payloadless compactor got unexpected new segment %d, want %d", segmentNum, activeSegmentNum+1)
+	}
+
+	prevSegmentNum := activeSegmentNum
+	activeSegmentNum = segmentNum
+
+	c.logger.Info().Msgf("finish writing segment file %v, payloadless trie update writing to segment %v; checkpoint triggers at segment %v",
+		prevSegmentNum, activeSegmentNum, nextCheckpointNum)
+
+	if nextCheckpointNum > prevSegmentNum {
+		return activeSegmentNum, -1, nil
+	}
+
+	// nextCheckpointNum == prevSegmentNum — enough segments accumulated.
+	tries := trieQueue.Tries()
+	return activeSegmentNum, nextCheckpointNum, tries
+}
+
 // latestV7CheckpointNum returns the highest V7 checkpoint number on disk,
 // or -1 if none exist or listing fails (with the error logged).
 func latestV7CheckpointNum(checkpointer *realWAL.Checkpointer, logger zerolog.Logger) int {