node: Make stream sync more resilient

core/node/events/stream.go

@@ -33,8 +33,12 @@ type Stream interface {
 }
 
 type SyncResultReceiver interface {
+	// OnUpdate is called each time a new cookie is available for a stream
 	OnUpdate(r *StreamAndCookie)
+	// OnSyncError is called when a sync subscription failed unrecoverable
 	OnSyncError(err error)
+	// OnStreamSyncDown is called when updates for a stream could not be given.
+	OnStreamSyncDown(StreamId)
 }
 
 // TODO: refactor interfaces.

core/node/events/stream_cache_test.go

@@ -199,6 +199,7 @@ func TestCacheEvictionWithFilledMiniBlockPool(t *testing.T) {
 type testStreamCacheViewEvictionSub struct {
 	receivedStreamAndCookies []*protocol.StreamAndCookie
 	receivedErrors           []error
+	streamErrors             []shared.StreamId
 }
 
 func (sub *testStreamCacheViewEvictionSub) OnUpdate(sac *protocol.StreamAndCookie) {
@@ -209,6 +210,10 @@ func (sub *testStreamCacheViewEvictionSub) OnSyncError(err error) {
 	sub.receivedErrors = append(sub.receivedErrors, err)
 }
 
+func (sub *testStreamCacheViewEvictionSub) OnStreamSyncDown(streamID shared.StreamId) {
+	sub.streamErrors = append(sub.streamErrors, streamID)
+}
+
 func (sub *testStreamCacheViewEvictionSub) eventsReceived() int {
 	count := 0
 	for _, sac := range sub.receivedStreamAndCookies {

core/node/rpc/info.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/river-build/river/core/node/rpc/sync"
 	"log/slog"
 	"strconv"
 
@@ -45,6 +46,7 @@ func (s *Service) info(
 ) (*connect.Response[InfoResponse], error) {
 	if len(request.Msg.Debug) > 0 {
 		debug := request.Msg.Debug[0]
+
 		if debug == "error" {
 			return nil, RiverError(Err_DEBUG_ERROR, "Error requested through Info request")
 		} else if debug == "network_error" {
@@ -54,6 +56,8 @@ func (s *Service) info(
 			return nil, errors.New("error requested through Info request")
 		} else if debug == "make_miniblock" {
 			return s.debugInfoMakeMiniblock(ctx, request)
+		} else if debug == "drop_stream" {
+			return s.debugDropStream(ctx, request)
 		}
 
 		if s.config.EnableTestAPIs {
@@ -83,6 +87,32 @@ func (s *Service) info(
 	}), nil
 }
 
+func (s *Service) debugDropStream(
+	ctx context.Context,
+	request *connect.Request[InfoRequest],
+) (*connect.Response[InfoResponse], error) {
+	if len(request.Msg.GetDebug()) < 3 {
+		return nil, RiverError(Err_DEBUG_ERROR, "drop_stream requires a sync id and stream id")
+	}
+
+	syncID := request.Msg.Debug[1]
+	streamID, err := shared.StreamIdFromString(request.Msg.Debug[2])
+	if err != nil {
+		return nil, err
+	}
+
+	dbgHandler, ok := s.syncHandler.(sync.DebugHandler)
+	if !ok {
+		return nil, RiverError(Err_UNAVAILABLE, "Drop stream not supported")
+	}
+
+	if err = dbgHandler.DebugDropStream(ctx, syncID, streamID); err != nil {
+		return nil, err
+	}
+
+	return connect.NewResponse(&InfoResponse{}), nil
+}
+
 func (s *Service) debugInfoMakeMiniblock(
 	ctx context.Context,
 	request *connect.Request[InfoRequest],

core/node/rpc/server.go

@@ -16,11 +16,6 @@ import (
 
 	"connectrpc.com/connect"
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/rs/cors"
-	"golang.org/x/net/http2"
-	"golang.org/x/net/http2/h2c"
-	httptrace "gopkg.in/DataDog/dd-trace-go.v1/contrib/net/http"
-
 	"github.com/river-build/river/core/config"
 	"github.com/river-build/river/core/node/auth"
 	. "github.com/river-build/river/core/node/base"
@@ -32,8 +27,13 @@ import (
 	. "github.com/river-build/river/core/node/protocol"
 	"github.com/river-build/river/core/node/protocol/protocolconnect"
 	"github.com/river-build/river/core/node/registries"
+	"github.com/river-build/river/core/node/rpc/sync"
 	"github.com/river-build/river/core/node/storage"
 	"github.com/river-build/river/core/xchain/entitlement"
+	"github.com/rs/cors"
+	"golang.org/x/net/http2"
+	"golang.org/x/net/http2/h2c"
+	httptrace "gopkg.in/DataDog/dd-trace-go.v1/contrib/net/http"
 )
 
 const (
@@ -504,11 +504,10 @@ func (s *Service) initCacheAndSync() error {
 
 	s.mbProducer = events.NewMiniblockProducer(s.serverCtx, s.cache, nil)
 
-	s.syncHandler = NewSyncHandler(
-		s.wallet,
+	s.syncHandler = sync.NewHandler(
+		s.wallet.Address,
 		s.cache,
 		s.nodeRegistry,
-		s.streamRegistry,
 	)
 
 	return nil

core/node/rpc/service.go

@@ -2,6 +2,7 @@ package rpc
 
 import (
 	"context"
+	river_sync "github.com/river-build/river/core/node/rpc/sync"
 	"log/slog"
 	"net"
 	"net/http"
@@ -44,7 +45,7 @@ type Service struct {
 	// Streams
 	cache       events.StreamCache
 	mbProducer  events.MiniblockProducer
-	syncHandler SyncHandler
+	syncHandler river_sync.Handler
 
 	// River chain
 	riverChain       *crypto.Blockchain

core/node/rpc/service_sync_streams.go

@@ -0,0 +1,55 @@
+package rpc
+
+import (
+	"connectrpc.com/connect"
+	"context"
+	. "github.com/river-build/river/core/node/protocol"
+)
+
+// TODO: wire metrics.
+// var (
+// 	syncStreamsRequests   = infra.NewSuccessMetrics("sync_streams_requests", serviceRequests)
+// 	syncStreamsResultSize = infra.NewCounter("sync_streams_result_size", "The total number of events returned by sync streams")
+// )
+
+// func addUpdatesToCounter(updates []*StreamAndCookie) {
+// 	for _, stream := range updates {
+// 		syncStreamsResultSize.Add(float64(len(stream.Events)))
+// 	}
+// }
+
+func (s *Service) SyncStreams(
+	ctx context.Context,
+	req *connect.Request[SyncStreamsRequest],
+	res *connect.ServerStream[SyncStreamsResponse],
+) error {
+	return s.syncHandler.SyncStreams(ctx, req, res)
+}
+
+func (s *Service) AddStreamToSync(
+	ctx context.Context,
+	req *connect.Request[AddStreamToSyncRequest],
+) (*connect.Response[AddStreamToSyncResponse], error) {
+	return s.syncHandler.AddStreamToSync(ctx, req)
+}
+
+func (s *Service) RemoveStreamFromSync(
+	ctx context.Context,
+	req *connect.Request[RemoveStreamFromSyncRequest],
+) (*connect.Response[RemoveStreamFromSyncResponse], error) {
+	return s.syncHandler.RemoveStreamFromSync(ctx, req)
+}
+
+func (s *Service) CancelSync(
+	ctx context.Context,
+	req *connect.Request[CancelSyncRequest],
+) (*connect.Response[CancelSyncResponse], error) {
+	return s.syncHandler.CancelSync(ctx, req)
+}
+
+func (s *Service) PingSync(
+	ctx context.Context,
+	req *connect.Request[PingSyncRequest],
+) (*connect.Response[PingSyncResponse], error) {
+	return s.syncHandler.PingSync(ctx, req)
+}