diff --git a/plugin/evm/atomic/sync/syncer.go b/plugin/evm/atomic/sync/syncer.go index 9d963f2899..f161ecfb70 100644 --- a/plugin/evm/atomic/sync/syncer.go +++ b/plugin/evm/atomic/sync/syncer.go @@ -146,6 +146,23 @@ func (s *Syncer) Sync(ctx context.Context) error { return s.syncer.Sync(ctx) } +// UpdateSyncTarget updates the target summary to sync to. +// TODO(alarso16): see https://github.com/ava-labs/coreth/issues/1143 +func (*Syncer) UpdateSyncTarget(message.Syncable) error { + return errors.New("not yet implemented") +} + +// Finalize is called after Sync completes to ensure all data matches the expected state. +// This may block after UpdateSyncTarget is implemented. +func (s *Syncer) Finalize(_ context.Context, summary message.Syncable) error { + // Ensure we synced to the expected target height. + if summary.Height() != s.targetHeight { + return fmt.Errorf("expected to sync to height %d but synced to %d", s.targetHeight, summary.Height()) + } + + return nil +} + // addZeroes returns the big-endian representation of `height`, prefixed with [common.HashLength] zeroes. func addZeroes(height uint64) []byte { // Key format is [height(8 bytes)][blockchainID(32 bytes)]. Start should be the diff --git a/plugin/evm/vmsync/doubles_test.go b/plugin/evm/vmsync/doubles_test.go index e0c6d3e030..743ab4f9da 100644 --- a/plugin/evm/vmsync/doubles_test.go +++ b/plugin/evm/vmsync/doubles_test.go @@ -9,9 +9,13 @@ import ( "sync" "time" + "github.com/ava-labs/coreth/plugin/evm/message" + syncpkg "github.com/ava-labs/coreth/sync" ) +var _ syncpkg.Syncer = FuncSyncer{} + // FuncSyncer adapts a function to the simple Syncer shape used in tests. It is // useful for defining small, behavior-driven syncers inline. type FuncSyncer struct { @@ -21,12 +25,14 @@ type FuncSyncer struct { // Sync calls the wrapped function and returns its result. func (f FuncSyncer) Sync(ctx context.Context) error { return f.fn(ctx) } +// Unnecessary for these tests. +func (FuncSyncer) UpdateSyncTarget(message.Syncable) error { return nil } +func (FuncSyncer) Finalize(context.Context, message.Syncable) error { return nil } + // Name returns the provided name or a default if unspecified. func (FuncSyncer) Name() string { return "Test Name" } func (FuncSyncer) ID() string { return "test_id" } -var _ syncpkg.Syncer = FuncSyncer{} - // NewBarrierSyncer returns a syncer that, upon entering Sync, calls wg.Done() to // signal it has started, then blocks until either: // - `releaseCh` is closed, returning nil; or diff --git a/plugin/evm/vmsync/registry.go b/plugin/evm/vmsync/registry.go index f6895807e3..1be980db17 100644 --- a/plugin/evm/vmsync/registry.go +++ b/plugin/evm/vmsync/registry.go @@ -7,6 +7,7 @@ import ( "context" "errors" "fmt" + "sync" "github.com/ava-labs/libevm/log" "golang.org/x/sync/errgroup" @@ -16,7 +17,10 @@ import ( synccommon "github.com/ava-labs/coreth/sync" ) -var errSyncerAlreadyRegistered = errors.New("syncer already registered") +var ( + errSyncerAlreadyRegistered = errors.New("syncer already registered") + errUpdateSyncTargetNotAllowed = errors.New("UpdateSyncTarget not allowed while syncers are not running") +) // SyncerTask represents a single syncer with its name for identification. type SyncerTask struct { @@ -28,6 +32,11 @@ type SyncerTask struct { type SyncerRegistry struct { syncers []SyncerTask registeredNames map[string]bool // Track registered IDs to prevent duplicates. + + // This lock protects access to target and updateTargetAllowed. + targetLock sync.Mutex + target message.Syncable + updateTargetAllowed bool // Only allowed while syncers are running. } // NewSyncerRegistry creates a new empty syncer registry. @@ -61,12 +70,12 @@ func (r *SyncerRegistry) RunSyncerTasks(ctx context.Context, summary message.Syn summaryBlockHashHex := summary.GetBlockHash().Hex() blockHeight := summary.Height() - g, ctx := errgroup.WithContext(ctx) + g, egCtx := errgroup.WithContext(ctx) for _, task := range r.syncers { g.Go(func() error { log.Info("starting syncer", "name", task.name, "summary", summaryBlockHashHex, "height", blockHeight) - if err := task.syncer.Sync(ctx); err != nil { + if err := task.syncer.Sync(egCtx); err != nil { log.Error("failed syncing", "name", task.name, "summary", summaryBlockHashHex, "height", blockHeight, "err", err) return fmt.Errorf("%s failed: %w", task.name, err) } @@ -76,11 +85,61 @@ func (r *SyncerRegistry) RunSyncerTasks(ctx context.Context, summary message.Syn }) } + // Allow UpdateSyncTarget calls while waiting for syncers to complete. + r.targetLock.Lock() + r.updateTargetAllowed = true + r.target = summary + r.targetLock.Unlock() + if err := g.Wait(); err != nil { return err } log.Info("all syncers completed successfully", "count", len(r.syncers), "summary", summaryBlockHashHex) + // At this point, UpdateSyncTarget should no longer be called, so we should prevent it. + r.targetLock.Lock() + r.updateTargetAllowed = false + finalTarget := r.target + r.targetLock.Unlock() + + // Finalize each syncer to ensure all data matches the expected state. + // Use a new error group to finalize in parallel. + g, egCtx = errgroup.WithContext(ctx) + for _, task := range r.syncers { + log.Info("finalizing syncer", "name", task.name, "summary", summaryBlockHashHex, "height", blockHeight) + g.Go(func() error { + if err := task.syncer.Finalize(egCtx, finalTarget); err != nil { + log.Error("failed finalizing", "name", task.name, "summary", summaryBlockHashHex, "height", blockHeight, "err", err) + return fmt.Errorf("%s finalize failed: %w", task.name, err) + } + return nil + }) + } + + return g.Wait() +} + +// UpdateSyncTarget updates the target for all registered syncers. +func (r *SyncerRegistry) UpdateSyncTarget(summary message.Syncable) error { + r.targetLock.Lock() + defer r.targetLock.Unlock() + + if !r.updateTargetAllowed { + return errUpdateSyncTargetNotAllowed + } + + // Since UpdateSyncTarget can block, we call each syncer in its own goroutine. + var eg errgroup.Group + for _, task := range r.syncers { + eg.Go(func() error { + return task.syncer.UpdateSyncTarget(summary) + }) + } + if err := eg.Wait(); err != nil { + return fmt.Errorf("failed to update sync target: %w", err) + } + + r.target = summary return nil } diff --git a/plugin/evm/vmsync/registry_test.go b/plugin/evm/vmsync/registry_test.go index 2a85661444..226f0f0be4 100644 --- a/plugin/evm/vmsync/registry_test.go +++ b/plugin/evm/vmsync/registry_test.go @@ -24,9 +24,11 @@ var _ syncpkg.Syncer = (*mockSyncer)(nil) // mockSyncer implements [syncpkg.Syncer] for testing. type mockSyncer struct { - name string - syncError error - started bool // Track if already started + name string + syncError error + started bool // Track if already started + updateCount int // Track number of UpdateSyncTarget calls + finalizeTarget message.Syncable } func newMockSyncer(name string, syncError error) *mockSyncer { @@ -38,6 +40,16 @@ func (m *mockSyncer) Sync(context.Context) error { return m.syncError } +func (m *mockSyncer) UpdateSyncTarget(message.Syncable) error { + m.updateCount++ + return nil +} + +func (m *mockSyncer) Finalize(_ context.Context, target message.Syncable) error { + m.finalizeTarget = target + return nil +} + func (m *mockSyncer) Name() string { return m.name } func (m *mockSyncer) ID() string { return m.name } @@ -47,9 +59,17 @@ type namedSyncer struct { syncer syncpkg.Syncer } -func (n *namedSyncer) Sync(ctx context.Context) error { return n.syncer.Sync(ctx) } func (n *namedSyncer) Name() string { return n.name } func (n *namedSyncer) ID() string { return n.name } +func (n *namedSyncer) Sync(ctx context.Context) error { return n.syncer.Sync(ctx) } + +func (n *namedSyncer) UpdateSyncTarget(target message.Syncable) error { + return n.syncer.UpdateSyncTarget(target) +} + +func (n *namedSyncer) Finalize(ctx context.Context, summary message.Syncable) error { + return n.syncer.Finalize(ctx, summary) +} // syncerConfig describes a test syncer setup for RunSyncerTasks table tests. type syncerConfig struct { @@ -307,6 +327,73 @@ func TestSyncerRegistry_NoSyncersRegistered(t *testing.T) { require.NoError(t, registry.RunSyncerTasks(ctx, newTestClientSummary(t))) } +func TestSyncerRegistry_UpdateSyncTarget(t *testing.T) { + t.Parallel() + + registry := NewSyncerRegistry() + ctx, cancel := utilstest.NewTestContext(t) + t.Cleanup(cancel) + + var allStartedWG sync.WaitGroup + allStartedWG.Add(1) + releaseCh := make(chan struct{}) + barrierSyncer := NewBarrierSyncer(&allStartedWG, releaseCh) + registry.Register(barrierSyncer) + + // Create a tracked syncer to verify UpdateSyncTarget calls + syncer := newMockSyncer("UpdateTargetSyncer", nil) + registry.Register(syncer) + + doneCh := make(chan error, 1) + go func() { doneCh <- registry.RunSyncerTasks(ctx, newTestClientSummary(t)) }() + + utilstest.WaitGroupWithTimeout(t, &allStartedWG, 2*time.Second, "timed out waiting for barrier syncers to start") + + customSummary, err := message.NewBlockSyncSummary(common.HexToHash("0xabcdef"), 1234, common.HexToHash("0xabcdef")) + require.NoError(t, err) + require.NoError(t, registry.UpdateSyncTarget(customSummary)) + close(releaseCh) + + require.NoError(t, utilstest.WaitErrWithTimeout(t, doneCh, 4*time.Second)) + require.Equal(t, 1, syncer.updateCount) + require.Equal(t, customSummary, syncer.finalizeTarget) +} + +func TestSyncerRegistry_UpdateSyncTargetNotAllowed(t *testing.T) { + t.Parallel() + + registry := NewSyncerRegistry() + ctx, cancel := utilstest.NewTestContext(t) + t.Cleanup(cancel) + + const numUpdateSyncers = 3 + var mockSyncers []*mockSyncer + for i := 0; i < numUpdateSyncers; i++ { + name := fmt.Sprintf("UpdateSyncer-%d", i) + mockSyncer := newMockSyncer(name, nil) + mockSyncers = append(mockSyncers, mockSyncer) + require.NoError(t, registry.Register(&namedSyncer{name: name, syncer: mockSyncer})) + } + + summary := newTestClientSummary(t) + + // Call UpdateSyncTarget before starting syncers + err := registry.UpdateSyncTarget(summary) + require.ErrorIs(t, err, errUpdateSyncTargetNotAllowed) + + // First call to RunSyncerTasks + require.NoError(t, registry.RunSyncerTasks(ctx, summary)) + + // Call UpdateSyncTarget after syncers have completed + err = registry.UpdateSyncTarget(summary) + require.ErrorIs(t, err, errUpdateSyncTargetNotAllowed) + + // Verify UpdateSyncTarget was not called for each syncer + for i, mockSyncer := range mockSyncers { + require.Zero(t, mockSyncer.updateCount, "UpdateSyncTarget was called for syncer %d", i) + } +} + func newTestClientSummary(t *testing.T) message.Syncable { t.Helper() summary, err := message.NewBlockSyncSummary(common.HexToHash("0xdeadbeef"), 1000, common.HexToHash("0xdeadbeef")) diff --git a/sync/blocksync/syncer.go b/sync/blocksync/syncer.go index ca69a6dabc..489c4df18b 100644 --- a/sync/blocksync/syncer.go +++ b/sync/blocksync/syncer.go @@ -13,6 +13,8 @@ import ( "github.com/ava-labs/libevm/ethdb" "github.com/ava-labs/libevm/log" + "github.com/ava-labs/coreth/plugin/evm/message" + syncpkg "github.com/ava-labs/coreth/sync" statesyncclient "github.com/ava-labs/coreth/sync/client" ) @@ -20,9 +22,10 @@ import ( const blocksPerRequest = 32 var ( - _ syncpkg.Syncer = (*BlockSyncer)(nil) - errBlocksToFetchRequired = errors.New("blocksToFetch must be > 0") - errFromHashRequired = errors.New("fromHash must be non-zero when fromHeight > 0") + _ syncpkg.Syncer = (*BlockSyncer)(nil) + + errBlocksToFetchRequired = errors.New("blocksToFetch must be > 0") + errFromHashRequired = errors.New("fromHash must be non-zero when fromHeight > 0") ) type BlockSyncer struct { @@ -112,3 +115,15 @@ func (s *BlockSyncer) Sync(ctx context.Context) error { log.Info("fetched blocks from peer", "total", blocksToFetch) return batch.Write() } + +// UpdateSyncTarget updates the target summary to sync to. +// TODO(alarso16): see https://github.com/ava-labs/coreth/issues/1260 +func (*BlockSyncer) UpdateSyncTarget(message.Syncable) error { + return errors.New("not yet implemented") +} + +// Finalize asserts that all blocks have been requested and written to disk. +// This is a no-op until UpdateSyncTarget is implemented. +func (*BlockSyncer) Finalize(context.Context, message.Syncable) error { + return nil +} diff --git a/sync/statesync/code_syncer.go b/sync/statesync/code_syncer.go index 3fdda1f6c9..4eb1db4f20 100644 --- a/sync/statesync/code_syncer.go +++ b/sync/statesync/code_syncer.go @@ -5,6 +5,7 @@ package statesync import ( "context" + "errors" "fmt" "sync" @@ -115,6 +116,17 @@ func (c *CodeSyncer) Sync(ctx context.Context) error { return eg.Wait() } +// UpdateSyncTarget is not yet implemented. +// TODO(alarso16): see https://github.com/ava-labs/coreth/issues/1341 +func (*CodeSyncer) UpdateSyncTarget(message.Syncable) error { + return errors.New("not yet implemented") +} + +// Finalize is currently a no-op since UpdateSyncTarget is not implemented. +func (*CodeSyncer) Finalize(context.Context, message.Syncable) error { + return nil +} + // work fulfills any incoming requests from the producer channel by fetching code bytes from the network // and fulfilling them by updating the database. func (c *CodeSyncer) work(ctx context.Context) error { diff --git a/sync/statesync/state_syncer.go b/sync/statesync/state_syncer.go index adcf1809d8..fffb967635 100644 --- a/sync/statesync/state_syncer.go +++ b/sync/statesync/state_syncer.go @@ -17,6 +17,7 @@ import ( "golang.org/x/sync/errgroup" "github.com/ava-labs/coreth/core/state/snapshot" + "github.com/ava-labs/coreth/plugin/evm/message" syncpkg "github.com/ava-labs/coreth/sync" syncclient "github.com/ava-labs/coreth/sync/client" @@ -30,9 +31,11 @@ const ( ) var ( - _ syncpkg.Syncer = (*stateSync)(nil) - errCodeRequestQueueRequired = errors.New("code request queue is required") - errLeafsRequestSizeRequired = errors.New("leafs request size must be > 0") + _ syncpkg.Syncer = (*stateSync)(nil) + + errCodeRequestQueueRequired = errors.New("code request queue is required") + errDynamicSyncerNotSupported = errors.New("dynamic syncer not supported") + errLeafsRequestSizeRequired = errors.New("leafs request size must be > 0") ) // stateSync keeps the state of the entire state sync operation. @@ -162,6 +165,23 @@ func (t *stateSync) Sync(ctx context.Context) error { return eg.Wait() } +// UpdateSyncTarget is not supported for this syncer. +func (*stateSync) UpdateSyncTarget(message.Syncable) error { + return errDynamicSyncerNotSupported +} + +// Finalize assures all data matches the expected state after a sync. +// This may block while resolving in-flight data. +func (t *stateSync) Finalize(_ context.Context, summary message.Syncable) error { + if t.root != summary.GetBlockRoot() { + return fmt.Errorf("finalizing sync with unexpected root: local %s, expected %s", t.root, summary.GetBlockRoot()) + } + + // The triedb should have the root available after syncing. + _, err := t.trieDB.Reader(t.root) + return err +} + // onStorageTrieFinished is called after a storage trie finishes syncing. func (t *stateSync) onStorageTrieFinished(root common.Hash) error { <-t.triesInProgressSem // allow another trie to start (release the semaphore) diff --git a/sync/types.go b/sync/types.go index d6e9005b06..7a07e10d56 100644 --- a/sync/types.go +++ b/sync/types.go @@ -22,6 +22,14 @@ type Syncer interface { // The sync will respect context cancellation. Sync(ctx context.Context) error + // UpdateSyncTarget updates the target summary to sync to. + // If this is not implemented on the syncer, it should return an error. + UpdateSyncTarget(summary message.Syncable) error + + // Finalize assures all data matches the expected state after a sync. + // This may block while resolving in-flight data. + Finalize(ctx context.Context, summary message.Syncable) error + // Name returns a human-readable name for this syncer implementation. Name() string