From ccdd500a0c3f63bc2879ddd89f5275322c5f5cac Mon Sep 17 00:00:00 2001 From: Thomas Dall'Agnese Date: Tue, 18 Aug 2026 21:12:30 +0100 Subject: [PATCH] Update proposal preparations when a consensus client becomes active. Proposal preparations are held in memory by beacon nodes, so a node that restarts loses them until vouch's next scheduled update, which runs once per epoch. A node that is inactive when that update runs is skipped, and not retried until the following epoch. A proposal in that window can be made with the beacon node's own default fee recipient rather than the configured one. Provide hooks to the consensus clients so that proposal preparations are updated as soon as a client becomes active, in addition to the existing per-epoch update. The updates use the service context rather than the one supplied to the hook. A client checks its connection state on the way in to a request that finds it inactive, so the hook can be called with that request's context, which is cancelled as soon as the request completes; using it cancels the preparations mid-flight. --- clients.go | 38 ++++++++++++ clients_test.go | 157 ++++++++++++++++++++++++++++++++++++++++++++++++ main.go | 23 +++++++ 3 files changed, 218 insertions(+) create mode 100644 clients_test.go diff --git a/clients.go b/clients.go index af812b13..10608979 100644 --- a/clients.go +++ b/clients.go @@ -33,6 +33,43 @@ var ( knownClientsMu sync.Mutex ) +var ( + reconnectCallbacks []func(ctx context.Context, address string) + reconnectCallbacksMu sync.RWMutex +) + +// addReconnectCallback adds a function that is called whenever a consensus client becomes +// active, either on first connection or on reconnection after the client has been +// unavailable. The client dispatches its hooks in a goroutine of their own, so callbacks +// never run on a request path; they are called one after another, so a slow callback delays +// those registered after it. +func addReconnectCallback(callback func(ctx context.Context, address string)) { + reconnectCallbacksMu.Lock() + defer reconnectCallbacksMu.Unlock() + + reconnectCallbacks = append(reconnectCallbacks, callback) +} + +// onClientActive calls the registered reconnect callbacks for the client at the given address. +func onClientActive(ctx context.Context, address string) { + reconnectCallbacksMu.RLock() + callbacks := make([]func(context.Context, string), len(reconnectCallbacks)) + copy(callbacks, reconnectCallbacks) + reconnectCallbacksMu.RUnlock() + + for _, callback := range callbacks { + callback(ctx, address) + } +} + +// clientHooks are the hooks provided to each consensus client, allowing vouch to react to +// changes in the client's connection state. +var clientHooks = &httpclient.Hooks{ + OnActive: func(ctx context.Context, s *httpclient.Service) { + onClientActive(ctx, s.Address()) + }, +} + // fetchClient fetches a client service, instantiating it if required. func fetchClient(ctx context.Context, monitor metrics.Service, address string) (eth2client.Service, error) { if address == "" { @@ -55,6 +92,7 @@ func fetchClient(ctx context.Context, monitor metrics.Service, address string) ( "User-Agent": fmt.Sprintf("Vouch/%s", ReleaseVersion), }), httpclient.WithReducedMemoryUsage(util.HierarchicalBool("reduced-memory-usage", fmt.Sprintf("eth2client.%s", address))), + httpclient.WithHooks(clientHooks), ) if err != nil { return nil, errors.Wrap(err, "failed to initiate consensus client") diff --git a/clients_test.go b/clients_test.go new file mode 100644 index 00000000..d124e989 --- /dev/null +++ b/clients_test.go @@ -0,0 +1,157 @@ +// Copyright © 2026 Attestant Limited. +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "context" + "testing" + + mockproposalpreparer "github.com/attestantio/vouch/services/proposalpreparer/mock" + "github.com/stretchr/testify/require" +) + +// resetReconnectCallbacks clears the registered callbacks, restoring them when the test ends. +func resetReconnectCallbacks(t *testing.T) { + t.Helper() + + reconnectCallbacksMu.Lock() + existing := reconnectCallbacks + reconnectCallbacks = nil + reconnectCallbacksMu.Unlock() + + t.Cleanup(func() { + reconnectCallbacksMu.Lock() + reconnectCallbacks = existing + reconnectCallbacksMu.Unlock() + }) +} + +func TestOnClientActiveWithoutCallbacks(t *testing.T) { + resetReconnectCallbacks(t) + + require.NotPanics(t, func() { + onClientActive(context.Background(), "http://localhost:5051") + }) +} + +func TestOnClientActiveCallsCallbacksWithAddress(t *testing.T) { + resetReconnectCallbacks(t) + + addresses := make([]string, 0) + addReconnectCallback(func(_ context.Context, address string) { + addresses = append(addresses, address) + }) + addReconnectCallback(func(_ context.Context, address string) { + addresses = append(addresses, address) + }) + + onClientActive(context.Background(), "http://localhost:5051") + + require.Equal(t, []string{"http://localhost:5051", "http://localhost:5051"}, addresses) +} + +func TestOnClientActiveCallsCallbacksOnEachActivation(t *testing.T) { + resetReconnectCallbacks(t) + + calls := 0 + addReconnectCallback(func(_ context.Context, _ string) { + calls++ + }) + + onClientActive(context.Background(), "http://localhost:5051") + onClientActive(context.Background(), "http://localhost:5052") + + require.Equal(t, 2, calls) +} + +// TestRegisterProposalPreparationsUpdaterWithoutPreparer confirms that a nil preparer, as +// provided before the bellatrix fork, does not register a callback that would panic when a +// client becomes active. +func TestRegisterProposalPreparationsUpdaterWithoutPreparer(t *testing.T) { + resetReconnectCallbacks(t) + + registerProposalPreparationsUpdater(context.Background(), nil) + + reconnectCallbacksMu.RLock() + registered := len(reconnectCallbacks) + reconnectCallbacksMu.RUnlock() + require.Equal(t, 0, registered) + + require.NotPanics(t, func() { + onClientActive(context.Background(), "http://localhost:5051") + }) +} + +// ctxCapturingProposalPreparer records the context with which it was last called. +type ctxCapturingProposalPreparer struct { + ctx context.Context +} + +func (s *ctxCapturingProposalPreparer) UpdatePreparations(ctx context.Context) error { + s.ctx = ctx + + return nil +} + +// TestRegisterProposalPreparationsUpdaterIgnoresCallbackContext confirms that the updates use the +// service context rather than the callback's. A client checks its connection state on the way in +// to a request that finds it inactive, so the callback can be invoked with that request's context, +// which is cancelled as soon as the request completes; using it cancels the preparations in flight. +func TestRegisterProposalPreparationsUpdaterIgnoresCallbackContext(t *testing.T) { + resetReconnectCallbacks(t) + + preparer := &ctxCapturingProposalPreparer{} + registerProposalPreparationsUpdater(context.Background(), preparer) + + // Fire the callback with a context that is already cancelled, as a completed request's is. + cancelledCtx, cancel := context.WithCancel(context.Background()) + cancel() + onClientActive(cancelledCtx, "http://localhost:5051") + + require.NotNil(t, preparer.ctx) + require.NoError(t, preparer.ctx.Err()) +} + +func TestRegisterProposalPreparationsUpdaterWithPreparer(t *testing.T) { + resetReconnectCallbacks(t) + + registerProposalPreparationsUpdater(context.Background(), mockproposalpreparer.New()) + + reconnectCallbacksMu.RLock() + registered := len(reconnectCallbacks) + reconnectCallbacksMu.RUnlock() + require.Equal(t, 1, registered) + + require.NotPanics(t, func() { + onClientActive(context.Background(), "http://localhost:5051") + }) +} + +// TestAddReconnectCallbackDuringDispatch confirms that a callback registered while callbacks +// are being dispatched does not deadlock, as clients can become active at any time. +func TestAddReconnectCallbackDuringDispatch(t *testing.T) { + resetReconnectCallbacks(t) + + addReconnectCallback(func(_ context.Context, _ string) { + addReconnectCallback(func(_ context.Context, _ string) {}) + }) + + require.NotPanics(t, func() { + onClientActive(context.Background(), "http://localhost:5051") + }) + + reconnectCallbacksMu.RLock() + defer reconnectCallbacksMu.RUnlock() + require.Len(t, reconnectCallbacks, 2) +} diff --git a/main.go b/main.go index 6a24835a..e297ef04 100644 --- a/main.go +++ b/main.go @@ -401,6 +401,7 @@ func startServices(ctx context.Context, if err != nil { return nil, nil, err } + registerProposalPreparationsUpdater(ctx, proposalPreparer) multiInstance, err := startMultiInstance(ctx, monitor, chainTime, eth2Client, beaconBlockHeaderProvider) if err != nil { @@ -505,6 +506,28 @@ func initController(ctx context.Context, return controller, nil } +// registerProposalPreparationsUpdater updates proposal preparations whenever a consensus client +// becomes active. Beacon nodes hold proposal preparations in memory, so a node that has restarted +// has none until the next scheduled update; this keeps that window as short as possible. +// +// The supplied context is used for the updates, rather than the context provided to the callback. +// A client checks its connection state on the way in to a request that finds it inactive, so the +// callback's context can be that request's, which is cancelled as soon as the request completes; +// using it here cancels the preparations mid-flight. +func registerProposalPreparationsUpdater(ctx context.Context, proposalPreparer proposalpreparer.Service) { + if proposalPreparer == nil { + // Not bellatrix-capable, so there are no preparations to provide. + return + } + + addReconnectCallback(func(_ context.Context, address string) { + log.Debug().Str("address", address).Msg("Consensus client active; updating proposal preparations") + if err := proposalPreparer.UpdatePreparations(ctx); err != nil { + log.Error().Str("address", address).Err(err).Msg("Failed to update proposal preparations on client activation") + } + }) +} + func initProposalPreparer(ctx context.Context, monitor metrics.Service, chainTime chaintime.Service, bellatrixCapable bool, accountManager accountmanager.Service, blockRelay blockrelay.Service) (proposalpreparer.Service, error) { // We need to submit proposal preparations to all nodes that are acting as beacon block proposers. nodeAddresses := util.BeaconNodeAddressesForProposing()