Proactive Syncer

internal/pebble/pebble.go

@@ -58,7 +58,7 @@ func NewDefaultConfig() Config {
 }
 
 // TODO: replace with AvalancheGo implementation (if still used for Vryx)
-func New(file string, cfg Config, registerer prometheus.Registerer) (*Database, error) {
+func New(dirName string, cfg Config, registerer prometheus.Registerer) (*Database, error) {
 	// These default settings are based on https://github.com/ethereum/go-ethereum/blob/master/ethdb/pebble/pebble.go
 	d := &Database{closing: make(chan struct{})}
 	opts := &pebble.Options{
@@ -96,7 +96,7 @@ func New(file string, cfg Config, registerer prometheus.Registerer) (*Database,
 		return nil, err
 	}
 	d.metrics = metrics
-	db, err := pebble.Open(file, opts)
+	db, err := pebble.Open(dirName, opts)
 	if err != nil {
 		return nil, err
 	}

internal/typedclient/client.go

@@ -0,0 +1,85 @@
+// Copyright (C) 2025, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package typedclient
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/network/p2p"
+	"github.com/ava-labs/avalanchego/snow/engine/common"
+	"github.com/ava-labs/avalanchego/utils"
+	"github.com/ava-labs/avalanchego/utils/set"
+)
+
+type Marshaler[T any, U any, V any] interface {
+	MarshalRequest(T) ([]byte, error)
+	UnmarshalResponse([]byte) (U, error)
+	MarshalGossip(V) ([]byte, error)
+}
+
+type TypedClient[T any, U any, V any] struct {
+	client    *p2p.Client
+	marshaler Marshaler[T, U, V]
+}
+
+func NewTypedClient[T any, U any, V any](client *p2p.Client, marshaler Marshaler[T, U, V]) *TypedClient[T, U, V] {
+	return &TypedClient[T, U, V]{
+		client:    client,
+		marshaler: marshaler,
+	}
+}
+
+func (t *TypedClient[T, U, _]) AppRequest(
+	ctx context.Context,
+	nodeID ids.NodeID,
+	request T,
+	onResponse func(ctx context.Context, nodeID ids.NodeID, response U, err error),
+) error {
+	onByteResponse := func(ctx context.Context, nodeID ids.NodeID, responseBytes []byte, err error) {
+		if err != nil {
+			onResponse(ctx, nodeID, utils.Zero[U](), err)
+			return
+		}
+
+		response, parseErr := t.marshaler.UnmarshalResponse(responseBytes)
+		if parseErr != nil {
+			onResponse(ctx, nodeID, utils.Zero[U](), parseErr)
+			return
+		}
+
+		onResponse(ctx, nodeID, response, err)
+	}
+
+	requestBytes, err := t.marshaler.MarshalRequest(request)
+	if err != nil {
+		return fmt.Errorf("failed to marshal request: %w", err)
+	}
+
+	return t.client.AppRequest(
+		ctx,
+		set.Of(nodeID),
+		requestBytes,
+		onByteResponse,
+	)
+}
+
+func (t *TypedClient[T, U, V]) AppGossip(
+	ctx context.Context,
+	gossip V,
+) error {
+	gossipBytes, err := t.marshaler.MarshalGossip(gossip)
+	if err != nil {
+		return fmt.Errorf("failed to marshal gossip: %w", err)
+	}
+
+	return t.client.AppGossip(
+		ctx,
+		common.SendConfig{
+			Validators: 100,
+		},
+		gossipBytes,
+	)
+}

internal/typedclient/sync_client.go

@@ -0,0 +1,60 @@
+// Copyright (C) 2024, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package typedclient
+
+import (
+	"context"
+
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/network/p2p"
+	"github.com/ava-labs/avalanchego/utils"
+)
+
+// SyncTypedClient provides a synchronous wrapper around TypedClient
+type SyncTypedClient[T any, U any, V any] struct {
+	client *TypedClient[T, U, V]
+}
+
+// NewSyncTypedClient creates a new synchronous client
+func NewSyncTypedClient[T any, U any, V any](client *p2p.Client, marshaler Marshaler[T, U, V]) *SyncTypedClient[T, U, V] {
+	return &SyncTypedClient[T, U, V]{
+		client: NewTypedClient(client, marshaler),
+	}
+}
+
+// SyncAppRequest sends a request and waits for the response synchronously
+func (s *SyncTypedClient[T, U, V]) SyncAppRequest(
+	ctx context.Context,
+	nodeID ids.NodeID,
+	request T,
+) (U, error) {
+	var (
+		response U
+		respErr  error
+		done     = make(chan struct{})
+	)
+
+	// We are guaranteed to eventually receive a response
+	onResponse := func(_ context.Context, _ ids.NodeID, resp U, err error) {
+		select {
+		case <-done:
+			return
+		default:
+			response = resp
+			respErr = err
+			close(done)
+		}
+	}
+
+	if err := s.client.AppRequest(ctx, nodeID, request, onResponse); err != nil {
+		return utils.Zero[U](), err
+	}
+
+	select {
+	case <-done:
+		return response, respErr
+	case <-ctx.Done():
+		return utils.Zero[U](), ctx.Err()
+	}
+}

internal/typedclient/sync_client_test.go

@@ -0,0 +1,166 @@
+// Copyright (C) 2024, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package typedclient
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/network/p2p/p2ptest"
+	"github.com/ava-labs/avalanchego/snow/engine/common"
+	"github.com/ava-labs/avalanchego/utils"
+	"github.com/stretchr/testify/require"
+)
+
+var _ Marshaler[testRequest, testResponse, []byte] = (*testMarshaler)(nil)
+
+func TestSyncAppRequest(t *testing.T) {
+	tests := []struct {
+		name                string
+		errResponse         bool
+		networkError        bool
+		expectedErrContains string
+		responseDelay       time.Duration
+		contextTimeout      time.Duration
+	}{
+		{
+			name:           "successful request",
+			contextTimeout: 2 * time.Second,
+		},
+		{
+			name:                "context timeout",
+			responseDelay:       1500 * time.Millisecond,
+			contextTimeout:      time.Second,
+			expectedErrContains: "context deadline exceeded",
+		},
+		{
+			name:                "error response",
+			contextTimeout:      2 * time.Second,
+			expectedErrContains: "simulated error",
+			errResponse:         true,
+		},
+	}
+
+	// Enforce run in parallel
+	t.Parallel()
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := require.New(t)
+			ctx, cancel := context.WithTimeout(context.Background(), tt.contextTimeout)
+			defer cancel()
+
+			handlerDone := make(chan struct{})
+			handler := &testHandler{
+				simulateError: tt.errResponse,
+				responseDelay: tt.responseDelay,
+				doneCh:        handlerDone,
+				response:      []byte(tt.name),
+			}
+
+			handlerID := ids.GenerateTestNodeID()
+
+			marshaler := testMarshaler{}
+			client := p2ptest.NewSelfClient(t, ctx, handlerID, handler)
+			syncClient := NewSyncTypedClient[testRequest, testResponse, []byte](client, &marshaler)
+
+			response, err := withContextEnforcement(ctx, tt.expectedErrContains, func() (testResponse, error) {
+				return syncClient.SyncAppRequest(ctx, handlerID, testRequest{message: "What is the current test name?"})
+			})
+
+			if len(tt.expectedErrContains) > 0 {
+				r.Error(err)
+				r.ErrorContains(err, tt.expectedErrContains)
+				r.Equal(utils.Zero[testResponse](), response)
+			} else {
+				r.NoError(err)
+				r.Equal(tt.name, response.reply)
+			}
+		})
+	}
+}
+
+// testHandler simulates different response behaviors
+type testHandler struct {
+	simulateError bool
+	responseDelay time.Duration
+	response      []byte
+	doneCh        chan struct{}
+}
+
+func (t *testHandler) Close() error {
+	if t.doneCh != nil {
+		close(t.doneCh)
+	}
+	return nil
+}
+
+func (t *testHandler) AppRequest(
+	ctx context.Context,
+	_ ids.NodeID,
+	_ time.Time,
+	_ []byte,
+) ([]byte, *common.AppError) {
+	if t.simulateError {
+		return nil, &common.AppError{Code: 1, Message: "simulated error"}
+	}
+
+	if t.responseDelay > 0 {
+		select {
+		case <-time.After(t.responseDelay):
+			// Delay completed
+		case <-ctx.Done():
+			// Context was canceled before delay completed
+			return nil, &common.AppError{
+				Code:    2,
+				Message: ctx.Err().Error(),
+			}
+		}
+	}
+
+	return t.response, nil
+}
+
+// AppGossip is required by the interface but not used in this test
+func (*testHandler) AppGossip(
+	_ context.Context,
+	_ ids.NodeID,
+	_ []byte,
+) {
+}
+
+type testRequest struct {
+	message string
+}
+
+type testResponse struct {
+	reply string
+}
+
+type testMarshaler struct{}
+
+func (testMarshaler) MarshalRequest(request testRequest) ([]byte, error) {
+	return []byte(request.message), nil
+}
+
+func (testMarshaler) UnmarshalResponse(bytes []byte) (testResponse, error) {
+	return testResponse{string(bytes)}, nil
+}
+
+func (testMarshaler) MarshalGossip(bytes []byte) ([]byte, error) {
+	return bytes, nil
+}
+
+// Ensure the context is cancelled before checking results
+func withContextEnforcement(ctx context.Context, expectedErr string, fn func() (testResponse, error)) (testResponse, error) {
+	// If we expect a context deadline error, just wait for the context to be cancelled
+	if expectedErr == "context deadline exceeded" {
+		<-ctx.Done()
+		return utils.Zero[testResponse](), ctx.Err()
+	}
+
+	// For other cases, run the function normally
+	return fn()
+}