[Storehouse] 016 - Payloadless Concurrent proof collection

cmd/execution_builder.go

@@ -941,16 +941,16 @@ func (exeNode *ExecutionNode) LoadExecutionStateLedger(
 	module.ReadyDoneAware,
 	error,
 ) {
+	// Ledger selection is two independent choices passed to the factory:
+	//   - --payloadless picks the payloadless vs. full ledger (this branch).
+	//   - --ledger-service-addr (Config.LedgerServiceAddr), when set, means this
+	//     node connects to a remote ledger service rather than running a local
+	//     ledger; the factory then returns a gRPC client instead of a local one.
+	// Combined: payloadless + remote address -> remote payloadless client;
+	// payloadless + no address -> local payloadless ledger; likewise for full mode.
 	if exeNode.exeConf.payloadless {
 		// Payloadless mode. ValidateFlags enforces --enable-storehouse,
 		// so the storehouse is the value source for reads.
-		//
-		// The factory call mirrors the full-mode call below: same Config,
-		// same triggerCheckpoint. Today the factory body is a placeholder
-		// (no WAL, no checkpoint load) — see TODOs at
-		// ledgerfactory.NewPayloadlessLedger. When the WAL/checkpoint
-		// pieces land, only the factory body changes; this call site stays
-		// the same.
 		pl, err := ledgerfactory.NewPayloadlessLedger(ledgerfactory.Config{
 			LedgerServiceAddr:     exeNode.exeConf.ledgerServiceAddr,
 			LedgerMaxRequestSize:  exeNode.exeConf.ledgerMaxRequestSize,
@@ -1485,11 +1485,52 @@ func (exeNode *ExecutionNode) LoadBootstrapper(node *NodeConfig) error {
 
 		// when bootstrapping, the bootstrap folder must have a checkpoint file
 		// we need to cover this file to the trie folder to restore the trie to restore the execution state.
+		//
+		// Note: in payloadless mode the V6 root checkpoint placed here is later
+		// converted to root.checkpoint.v7 by ledgerfactory.NewPayloadlessLedger
+		// before the bundle reads it. Bootstrap itself stays mode-agnostic.
 		err = copyBootstrapState(node.BootstrapDir, exeNode.exeConf.triedir)
 		if err != nil {
 			return fmt.Errorf("could not load bootstrap state from checkpoint file: %w", err)
 		}
 
+		// In payloadless (V7) mode the spork only produces a V6 root.checkpoint.
+		// Convert it to a V7 root checkpoint here so the payloadless ledger can
+		// seed its forest from it on first boot; later restarts reuse this file
+		// (or a newer numbered V7 checkpoint written by the compactor). The
+		// HasRootCheckpointV7 guard keeps a re-entry after an interrupted
+		// bootstrap from hitting ConvertCheckpointV6ToV7's "output exists" check.
+		//
+		// Only nodes running a local payloadless ledger need this: a node using a
+		// remote ledger service (ledgerServiceAddr set) never reads its local trie
+		// dir, and the remote ledger service performs its own V7 bootstrap. Skipping
+		// the conversion avoids a needless full-forest load on remote-ledger nodes.
+		//
+		// TODO: ConvertCheckpointV6ToV7 reads the entire V6 forest into memory
+		// before emitting V7, a memory/time spike at first boot for mainnet-scale
+		// root checkpoints. A future optimization is to convert subtrie-by-subtrie
+		// without loading the whole forest.
+		if exeNode.exeConf.payloadless && exeNode.exeConf.ledgerServiceAddr == "" {
+			triedir := exeNode.exeConf.triedir
+			hasV7Root, err := wal.HasRootCheckpointV7(triedir)
+			if err != nil {
+				return fmt.Errorf("could not check for V7 root checkpoint: %w", err)
+			}
+			if !hasV7Root {
+				err = wal.ConvertCheckpointV6ToV7(
+					triedir,
+					modelbootstrap.FilenameWALRootCheckpoint,
+					triedir,
+					modelbootstrap.FilenameWALRootCheckpoint+wal.V7FileSuffix,
+					node.Logger,
+					16,
+				)
+				if err != nil {
+					return fmt.Errorf("could not convert V6 root checkpoint to V7 for payloadless node: %w", err)
+				}
+			}
+		}
+
 		err = bootstrapper.BootstrapExecutionDatabase(node.StorageLockMgr, node.ProtocolDB, node.RootSeal)
 		if err != nil {
 			return fmt.Errorf("could not bootstrap execution database: %w", err)

cmd/ledger/main.go

@@ -18,9 +18,11 @@ import (
 	"go.uber.org/atomic"
 	"google.golang.org/grpc"
 
+	"github.com/onflow/flow-go/ledger"
 	ledgerfactory "github.com/onflow/flow-go/ledger/factory"
 	ledgerpb "github.com/onflow/flow-go/ledger/protobuf"
 	"github.com/onflow/flow-go/ledger/remote"
+	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/irrecoverable"
 	"github.com/onflow/flow-go/module/metrics"
 )
@@ -35,6 +37,7 @@ var (
 	checkpointDist      = flag.Uint("checkpoint-distance", 100, "Checkpoint distance")
 	checkpointsToKeep   = flag.Uint("checkpoints-to-keep", 3, "Number of checkpoints to keep")
 	logLevel            = flag.String("loglevel", "info", "Log level (panic, fatal, error, warn, info, debug)")
+	payloadless         = flag.Bool("payloadless", false, "Run the ledger service in payloadless mode (stores leaf hashes instead of full payloads; requires a V7 checkpoint in --triedir).")
 	maxRequestSize      = flag.Uint("max-request-size", 1<<30, "Maximum request message size in bytes (default: 1 GiB)")
 	maxResponseSize     = flag.Uint("max-response-size", 1<<30, "Maximum response message size in bytes (default: 1 GiB)")
 )
@@ -72,14 +75,18 @@ func main() {
 		Str("admin_addr", *adminAddr).
 		Uint("metrics_port", *metricsPort).
 		Int("mtrie_cache_size", *mtrieCacheSize).
+		Bool("payloadless", *payloadless).
 		Msg("starting ledger service")
 
 	// Create trigger for manual checkpointing (used by admin command)
 	triggerCheckpointOnNextSegmentFinish := atomic.NewBool(false)
 
-	// Create ledger using factory
+	// Create ledger using factory. The same config drives both modes; the
+	// payloadless flag selects which factory constructor (and gRPC service) is
+	// wired up. A ledger gRPC server registers either the full [remote.Service]
+	// or the [remote.PayloadlessService], never both.
 	metricsCollector := metrics.NewLedgerCollector("ledger", "wal")
-	ledgerStorage, err := ledgerfactory.NewLedger(ledgerfactory.Config{
+	factoryConfig := ledgerfactory.Config{
 		Triedir:            *triedir,
 		MTrieCacheSize:     uint32(*mtrieCacheSize),
 		CheckpointDistance: *checkpointDist,
@@ -88,24 +95,58 @@ func main() {
 		WALMetrics:         metricsCollector,
 		LedgerMetrics:      metricsCollector,
 		Logger:             logger,
-	}, triggerCheckpointOnNextSegmentFinish)
-	if err != nil {
-		logger.Fatal().Err(err).Msg("failed to create ledger")
+	}
+
+	// ledgerStorage is the lifecycle handle used for readiness, health check,
+	// and shutdown regardless of mode. registerService binds the mode-specific
+	// gRPC service onto the server once it is created.
+	var ledgerStorage module.ReadyDoneAware
+	var registerService func(grpcServer *grpc.Server)
+
+	if *payloadless {
+		payloadlessLedger, err := ledgerfactory.NewPayloadlessLedger(factoryConfig, triggerCheckpointOnNextSegmentFinish)
+		if err != nil {
+			logger.Fatal().Err(err).Msg("failed to create payloadless ledger")
+		}
+		ledgerStorage = payloadlessLedger
+		registerService = func(grpcServer *grpc.Server) {
+			ledgerpb.RegisterPayloadlessLedgerServiceServer(grpcServer, remote.NewPayloadlessService(payloadlessLedger, logger))
+		}
+	} else {
+		fullLedger, err := ledgerfactory.NewLedger(factoryConfig, triggerCheckpointOnNextSegmentFinish)
+		if err != nil {
+			logger.Fatal().Err(err).Msg("failed to create ledger")
+		}
+		ledgerStorage = fullLedger
+		registerService = func(grpcServer *grpc.Server) {
+			ledgerpb.RegisterLedgerServiceServer(grpcServer, remote.NewService(fullLedger, logger))
+		}
 	}
 
 	// Wait for ledger to be ready (WAL replay)
 	logger.Info().Msg("waiting for ledger initialization...")
 	<-ledgerStorage.Ready()
 	logger.Info().Msg("ledger ready")
 
+	// Both the full and payloadless ledgers expose state inspection for the
+	// post-startup health check, though only the full ledger declares it on its
+	// public interface; assert it here so the check works in either mode.
+	inspector, ok := ledgerStorage.(interface {
+		StateCount() int
+		StateByIndex(index int) (ledger.State, error)
+	})
+	if !ok {
+		logger.Fatal().Msg("ledger does not support state inspection")
+	}
+
 	// Check if any trie is loaded after startup
-	stateCount := ledgerStorage.StateCount()
+	stateCount := inspector.StateCount()
 	if stateCount == 0 {
 		logger.Fatal().Msg("no trie loaded after startup - no states available")
 	}
 
 	// Get the last trie state for logging
-	lastState, err := ledgerStorage.StateByIndex(-1)
+	lastState, err := inspector.StateByIndex(-1)
 	if err != nil {
 		logger.Fatal().Err(err).Msg("failed to get last state for logging")
 	}
@@ -123,9 +164,8 @@ func main() {
 		grpc.MaxSendMsgSize(int(*maxResponseSize)),
 	)
 
-	// Create and register ledger service
-	ledgerService := remote.NewService(ledgerStorage, logger)
-	ledgerpb.RegisterLedgerServiceServer(grpcServer, ledgerService)
+	// Register the mode-specific ledger service
+	registerService(grpcServer)
 
 	// Create listeners based on provided flags
 	type listenerInfo struct {

cmd/util/cmd/checkpoint-collect-stats/cmd.go

@@ -3,6 +3,7 @@ package checkpoint_collect_stats
 import (
 	"cmp"
 	"encoding/hex"
+	"fmt"
 	"math"
 	"slices"
 	"strings"
@@ -315,6 +316,15 @@ func getPayloadStatsFromCheckpoint(payloadCallBack func(payload *ledger.Payload)
 	memAllocBefore := debug.GetHeapAllocsBytes()
 	log.Info().Msgf("loading checkpoint(s) from %v", flagCheckpointDir)
 
+	// checkpoint-collect-stats analyzes payload contents (register types, sizes,
+	// account info). V7 (payloadless) checkpoints store only leaf hashes and contain
+	// no payloads, so they cannot be processed here. The WAL replay below loads only
+	// V6 checkpoints and silently ignores V7 files, which would otherwise produce
+	// misleading (stale or empty) stats. Fail fast with a clear error instead.
+	if err := requireV6Checkpoint(flagCheckpointDir); err != nil {
+		log.Fatal().Err(err).Msg("cannot collect stats from checkpoint")
+	}
+
 	diskWal, err := wal.NewDiskWAL(zerolog.Nop(), nil, &metrics.NoopCollector{}, flagCheckpointDir, complete.DefaultCacheSize, pathfinder.PathByteSize, wal.SegmentSize)
 	if err != nil {
 		log.Fatal().Err(err).Msg("cannot create WAL")
@@ -369,6 +379,33 @@ func getPayloadStatsFromCheckpoint(payloadCallBack func(payload *ledger.Payload)
 	return ledgerStats
 }
 
+// requireV6Checkpoint returns an error if the latest checkpoint in dir is a V7
+// (payloadless) checkpoint. checkpoint-collect-stats requires full payloads,
+// which V7 checkpoints do not contain.
+//
+// Only numbered checkpoints are considered (the WAL bootstrap loads the latest
+// numbered V6 checkpoint). If the latest numbered checkpoint is V7, this command
+// would otherwise silently fall back to an older V6 checkpoint or an empty state,
+// reporting misleading stats.
+//
+// Expected error returns during normal operation:
+//   - an error when the latest checkpoint in dir is a V7 (payloadless) checkpoint
+func requireV6Checkpoint(dir string) error {
+	_, latest, err := wal.ListCheckpointsWithInfo(dir)
+	if err != nil {
+		return fmt.Errorf("cannot list checkpoints in %s: %w", dir, err)
+	}
+
+	if latest != nil && latest.Version == wal.VersionV7 {
+		return fmt.Errorf(
+			"checkpoint %d in %s is a V7 (payloadless) checkpoint, which contains no payloads; "+
+				"checkpoint-collect-stats requires a V6 checkpoint",
+			latest.Number, dir)
+	}
+
+	return nil
+}
+
 func getRegisterStats(valueSizesByType sizesByType) []RegisterStatsByTypes {
 	domainStats := make([]RegisterStatsByTypes, 0, len(common.AllStorageDomains))
 	var allDomainSizes []float64

cmd/util/cmd/checkpoint-collect-stats/cmd_test.go

@@ -0,0 +1,56 @@
+package checkpoint_collect_stats
+
+import (
+	"testing"
+
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
+
+	"github.com/onflow/flow-go/ledger"
+	"github.com/onflow/flow-go/ledger/common/testutils"
+	"github.com/onflow/flow-go/ledger/complete/mtrie/trie"
+	"github.com/onflow/flow-go/ledger/complete/payloadless"
+	"github.com/onflow/flow-go/ledger/complete/wal"
+)
+
+// TestRequireV6Checkpoint_EmptyDir verifies that a directory without any numbered
+// checkpoint is accepted (the caller proceeds with WAL replay / root checkpoint).
+func TestRequireV6Checkpoint_EmptyDir(t *testing.T) {
+	require.NoError(t, requireV6Checkpoint(t.TempDir()))
+}
+
+// TestRequireV6Checkpoint_V6 verifies that a directory whose latest checkpoint is
+// V6 is accepted.
+func TestRequireV6Checkpoint_V6(t *testing.T) {
+	dir := t.TempDir()
+
+	p := testutils.PathByUint8(0)
+	v := testutils.LightPayload8('A', 'a')
+	tr, _, err := trie.NewTrieWithUpdatedRegisters(
+		trie.NewEmptyMTrie(), []ledger.Path{p}, []ledger.Payload{*v}, true)
+	require.NoError(t, err)
+
+	require.NoError(t, wal.StoreCheckpointV6Concurrently(
+		[]*trie.MTrie{tr}, dir, wal.NumberToFilename(1), zerolog.Nop()))
+
+	require.NoError(t, requireV6Checkpoint(dir))
+}
+
+// TestRequireV6Checkpoint_V7 verifies that a directory whose latest checkpoint is
+// V7 (payloadless) is rejected, since this command requires full payloads.
+func TestRequireV6Checkpoint_V7(t *testing.T) {
+	dir := t.TempDir()
+
+	p := testutils.PathByUint8(0)
+	v := testutils.LightPayload8('A', 'a')
+	tr, _, err := payloadless.NewTrieWithUpdatedRegisters(
+		payloadless.NewEmptyMTrie(), []ledger.Path{p}, [][]byte{v.Value()}, true)
+	require.NoError(t, err)
+
+	require.NoError(t, wal.StoreCheckpointV7Concurrently(
+		[]*payloadless.MTrie{tr}, dir, wal.NumberToFilenameV7(1), zerolog.Nop()))
+
+	err = requireV6Checkpoint(dir)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "V7")
+}