Update proposal preparations when a consensus client becomes active

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")

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)
+}

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()