diff --git a/internal/api/syncapi/cmdstreamutil.go b/internal/api/syncapi/cmdstreamutil.go index bbf4ab15..c4f52340 100644 --- a/internal/api/syncapi/cmdstreamutil.go +++ b/internal/api/syncapi/cmdstreamutil.go @@ -49,9 +49,6 @@ func (s *bidiSyncCommandStream) Send(item *v1sync.SyncStreamItem) { // SendErrorAndTerminate sends an error to the termination channel. // If the error is nil, it terminates only. func (s *bidiSyncCommandStream) SendErrorAndTerminate(err error) { - if err == nil { - return - } select { case s.terminateWithErrChan <- err: default: diff --git a/internal/api/syncapi/handler.nogo b/internal/api/syncapi/handler.nogo deleted file mode 100644 index 64bc711c..00000000 --- a/internal/api/syncapi/handler.nogo +++ /dev/null @@ -1,288 +0,0 @@ -package syncapi - -import ( - "context" - "errors" - "fmt" - "io" - "sync" - "time" - "unique" - - "connectrpc.com/connect" - "github.com/garethgeorge/backrest/gen/go/types" - v1 "github.com/garethgeorge/backrest/gen/go/v1" - "github.com/garethgeorge/backrest/gen/go/v1sync" - "github.com/garethgeorge/backrest/gen/go/v1sync/v1syncconnect" - "github.com/garethgeorge/backrest/internal/logstore" - "github.com/garethgeorge/backrest/internal/oplog" - "github.com/garethgeorge/backrest/internal/protoutil" - lru "github.com/hashicorp/golang-lru/v2" - "go.uber.org/zap" - "google.golang.org/protobuf/proto" - "google.golang.org/protobuf/types/known/emptypb" -) - -type opIdCacheKey struct { - OriginalInstanceKeyid unique.Handle[string] - ID int64 -} - -// syncHandler provides server component functionality for the sync API. -type syncHandler struct { - v1syncconnect.UnimplementedSyncPeerServiceHandler - oplog *oplog.OpLog - logStore *logstore.LogStore - - opCacheMu sync.Mutex - flowIDLru *lru.Cache[opIdCacheKey, int64] - opIDLru *lru.Cache[opIdCacheKey, int64] - - logger *zap.Logger -} - -func NewSyncHandler(oplog *oplog.OpLog, logStore *logstore.LogStore) *syncHandler { - // Both caches want to be reasonably large to avoid db lookups. - flowIDLru, _ := lru.New[opIdCacheKey, int64](4 * 1024) - opIDLru, _ := lru.New[opIdCacheKey, int64](16 * 1024) - - return &syncHandler{ - oplog: oplog, - logStore: logStore, - flowIDLru: flowIDLru, - opIDLru: opIDLru, - logger: zap.NewNop(), - } -} - -var _ v1syncconnect.SyncPeerServiceHandler = (*syncHandler)(nil) - -func (sh *syncHandler) SetLogger(logger *zap.Logger) *syncHandler { - if logger == nil { - logger = zap.NewNop() - } - sh.logger = logger.Named("synchandler") - return sh -} - -// translateSingleID translates a single ID (either opID or flowID) using the provided cache and query -func (sh *syncHandler) translateSingleID( - originalInstanceKeyid string, - originalID int64, - cache *lru.Cache[opIdCacheKey, int64], - query oplog.Query, -) (int64, error) { - if originalID == 0 { - return 0, nil - } - - cacheKey := opIdCacheKey{ - OriginalInstanceKeyid: unique.Make(originalInstanceKeyid), - ID: originalID, - } - - // Check cache first - if translatedID, ok := cache.Get(cacheKey); ok { - return translatedID, nil - } - - // Cache miss - query the database - op, err := sh.oplog.FindOneMetadata(query) - if err != nil { - if errors.Is(err, oplog.ErrNoResults) { - return 0, nil // No results means the ID is not found - } - return 0, err // Other errors should be propagated - } - - // Cache the result and return - translatedID := op.FlowID - cache.Add(cacheKey, translatedID) - return translatedID, nil -} - -func (sh *syncHandler) translateOpIdAndFlowID(originalInstanceKeyid string, originalOpId int64, originalFlowId int64) (int64, int64, error) { - sh.opCacheMu.Lock() - defer sh.opCacheMu.Unlock() - - // Translate opID - opID, err := sh.translateSingleID( - originalInstanceKeyid, - originalOpId, - sh.opIDLru, - oplog.Query{ - OriginalInstanceKeyid: &originalInstanceKeyid, - OriginalID: &originalOpId, - }, - ) - if err != nil { - return 0, 0, err - } - - // Translate flowID - flowID, err := sh.translateSingleID( - originalInstanceKeyid, - originalFlowId, - sh.flowIDLru, - oplog.Query{ - OriginalInstanceKeyid: &originalInstanceKeyid, - OriginalFlowID: &originalFlowId, - }, - ) - if err != nil { - return 0, 0, err - } - - return opID, flowID, nil -} - -func (sh *syncHandler) GetOperationMetadata(ctx context.Context, req *connect.Request[v1.OpSelector]) (*connect.Response[v1sync.GetOperationMetadataResponse], error) { - peer := PeerFromContext(ctx) - if peer == nil { - return nil, connect.NewError(connect.CodeUnauthenticated, errors.New("no peer found in context")) - } - - // Check if the peer can read the selector - if req.Msg.OriginalInstanceKeyid == nil || *req.Msg.OriginalInstanceKeyid != peer.Keyid { - return nil, connect.NewError(connect.CodePermissionDenied, errors.New("GetOperationMetadata: peer must specify original instance keyid")) - } - - sel, err := protoutil.OpSelectorToQuery(req.Msg) - if err != nil { - return nil, connect.NewError(connect.CodeInvalidArgument, err) - } - - var opIDs []int64 - var modNos []int64 - - if err := sh.oplog.QueryMetadata(sel, func(op oplog.OpMetadata) error { - opIDs = append(opIDs, op.OriginalID) - modNos = append(modNos, op.Modno) - return nil - }); err != nil { - return nil, connect.NewError(connect.CodeInternal, err) - } - - return connect.NewResponse(&v1sync.GetOperationMetadataResponse{ - OpIds: opIDs, - Modnos: modNos, - }), nil -} - -func (sh *syncHandler) SendOperations(ctx context.Context, stream *connect.ClientStream[v1.Operation]) (*connect.Response[emptypb.Empty], error) { - peer := PeerFromContext(ctx) - if peer == nil { - return nil, connect.NewError(connect.CodeUnauthenticated, errors.New("no peer found in context")) - } - - for stream.Receive() { - op := stream.Msg() - if op == nil { - return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("received nil operation")) - } - - id, flowID, err := sh.translateOpIdAndFlowID(peer.Keyid, op.Id, op.FlowId) - if err != nil { - return nil, connect.NewError(connect.CodeInternal, err) - } - - // Update the operation with the translated IDs - opCopy := proto.Clone(op).(*v1.Operation) - opCopy.Id = id - opCopy.FlowId = flowID - - // Set the operation in the oplog - if err := sh.oplog.Set(opCopy); err != nil { - return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("set operation: %w", err)) - } - } - - return nil, connect.NewError(connect.CodeUnimplemented, errors.New("v1sync.SyncPeerService.SendOperations is not implemented")) -} - -func (sh *syncHandler) GetLog(ctx context.Context, req *connect.Request[types.StringValue], stream *connect.ServerStream[v1sync.LogDataEntry]) error { - peer := PeerFromContext(ctx) - if peer == nil { - return connect.NewError(connect.CodeUnauthenticated, errors.New("no peer found in context")) - } - logID := req.Msg.Value - - metadata, err := sh.logStore.GetMetadata(logID) - if err != nil { - if errors.Is(err, logstore.ErrLogNotFound) { - return connect.NewError(connect.CodeNotFound, fmt.Errorf("log with ID %s not found", logID)) - } - return connect.NewError(connect.CodeInternal, fmt.Errorf("get log metadata: %w", err)) - } - - log, err := sh.logStore.Open(logID) - if err != nil { - return connect.NewError(connect.CodeInternal, fmt.Errorf("get log: %w", err)) - } else if log == nil { - return connect.NewError(connect.CodeNotFound, fmt.Errorf("log with ID %s not found", logID)) - } - defer log.Close() - - // Send first entry with log ID and owner operation ID - entry := &v1sync.LogDataEntry{ - LogId: logID, - OwnerOpid: metadata.OwnerOpID, - } - if metadata.ExpirationTime != (time.Time{}) { - entry.ExpirationTsUnix = metadata.ExpirationTime.Unix() - } - if err := stream.Send(entry); err != nil { - if errors.Is(err, io.EOF) { - return nil // Client closed the stream - } - return connect.NewError(connect.CodeInternal, fmt.Errorf("send log entry: %w", err)) - } - - // Read the log in chunks and send each chunk as a LogDataEntry - buf := make([]byte, 0, 32*1024) - for { - n, err := log.Read(buf) - if err != nil { - if errors.Is(err, io.EOF) { - break // End of log - } - return connect.NewError(connect.CodeInternal, fmt.Errorf("read log: %w", err)) - } - if n == 0 { - break - } - bytes := buf[:n] - entry := &v1sync.LogDataEntry{ - Chunk: bytes, - } - if err := stream.Send(entry); err != nil { - if errors.Is(err, io.EOF) { - break // Client closed the stream - } - return connect.NewError(connect.CodeInternal, fmt.Errorf("send log entry: %w", err)) - } - } - if err := log.Close(); err != nil { - return connect.NewError(connect.CodeInternal, fmt.Errorf("close log: %w", err)) - } - return nil -} - -func (sh *syncHandler) SetAvailableResources(ctx context.Context, req *connect.Request[v1sync.SetAvailableResourcesRequest]) (*connect.Response[emptypb.Empty], error) { - peer := PeerFromContext(ctx) - if peer == nil { - return nil, connect.NewError(connect.CodeUnauthenticated, errors.New("no peer found in context")) - } - return nil, connect.NewError(connect.CodeUnimplemented, errors.New("v1sync.SyncPeerService.SetAvailableResources is not implemented")) -} - -func (sh *syncHandler) SetConfig(context.Context, *connect.Request[v1sync.SetConfigRequest]) (*connect.Response[emptypb.Empty], error) { - return nil, connect.NewError(connect.CodeUnimplemented, errors.New("v1sync.SyncPeerService.SetConfig is not implemented")) -} - -func (sh *syncHandler) GetConfig(context.Context, *connect.Request[emptypb.Empty]) (*connect.Response[v1sync.RemoteConfig], error) { - return nil, connect.NewError(connect.CodeUnimplemented, errors.New("v1sync.SyncPeerService.GetConfig is not implemented")) -} - -type syncHandlerClient struct { -} diff --git a/internal/api/syncapi/syncapi_test.nogo b/internal/api/syncapi/syncapi_test.go similarity index 96% rename from internal/api/syncapi/syncapi_test.nogo rename to internal/api/syncapi/syncapi_test.go index 606dec0b..8bd8d2e1 100644 --- a/internal/api/syncapi/syncapi_test.nogo +++ b/internal/api/syncapi/syncapi_test.go @@ -13,9 +13,10 @@ import ( "time" v1 "github.com/garethgeorge/backrest/gen/go/v1" - "github.com/garethgeorge/backrest/gen/go/v1/v1connect" "github.com/garethgeorge/backrest/gen/go/v1sync" + "github.com/garethgeorge/backrest/gen/go/v1sync/v1syncconnect" "github.com/garethgeorge/backrest/internal/config" + "github.com/garethgeorge/backrest/internal/config/migrations" "github.com/garethgeorge/backrest/internal/cryptoutil" "github.com/garethgeorge/backrest/internal/logstore" "github.com/garethgeorge/backrest/internal/oplog" @@ -71,7 +72,6 @@ var ( ) func TestConnectionSucceeds(t *testing.T) { - t.Skip("skipping syncapi test") testutil.InstallZapLogger(t) ctx, _ := context.WithTimeout(context.Background(), 5*time.Second) @@ -79,6 +79,7 @@ func TestConnectionSucceeds(t *testing.T) { peerClientAddr := testutil.AllocOpenBindAddr(t) peerHostConfig := &v1.Config{ + Version: migrations.CurrentVersion, Instance: defaultHostID, Repos: []*v1.Repo{}, Multihost: &v1.Multihost{ @@ -94,6 +95,7 @@ func TestConnectionSucceeds(t *testing.T) { } peerClientConfig := &v1.Config{ + Version: migrations.CurrentVersion, Instance: defaultClientID, Repos: []*v1.Repo{}, Multihost: &v1.Multihost{ @@ -118,7 +120,6 @@ func TestConnectionSucceeds(t *testing.T) { } func TestConnectionBadKeyRejected(t *testing.T) { - t.Skip("skipping syncapi test") testutil.InstallZapLogger(t) ctx, _ := context.WithTimeout(context.Background(), 5*time.Second) @@ -127,6 +128,7 @@ func TestConnectionBadKeyRejected(t *testing.T) { // Host has identity1, and authorizes no one. peerHostConfig := &v1.Config{ + Version: migrations.CurrentVersion, Instance: defaultHostID, Repos: []*v1.Repo{}, Multihost: &v1.Multihost{ @@ -137,6 +139,7 @@ func TestConnectionBadKeyRejected(t *testing.T) { // Client has identity2 and tries to connect to host. peerClientConfig := &v1.Config{ + Version: migrations.CurrentVersion, Instance: defaultClientID, Repos: []*v1.Repo{}, Multihost: &v1.Multihost{ @@ -161,7 +164,6 @@ func TestConnectionBadKeyRejected(t *testing.T) { } func TestSyncConfigChange(t *testing.T) { - t.Skip("skipping syncapi test") testutil.InstallZapLogger(t) ctx, _ := context.WithTimeout(context.Background(), 10*time.Second) @@ -169,13 +171,16 @@ func TestSyncConfigChange(t *testing.T) { peerClientAddr := testutil.AllocOpenBindAddr(t) peerHostConfig := &v1.Config{ + Version: migrations.CurrentVersion, Instance: defaultHostID, Repos: []*v1.Repo{ { + Uri: "test-uri-should-sync", Id: defaultRepoID, Guid: defaultRepoGUID, }, { + Uri: "test-uri-do-not-sync", Id: "do-not-sync", Guid: cryptoutil.MustRandomID(cryptoutil.DefaultIDBits), }, @@ -201,6 +206,7 @@ func TestSyncConfigChange(t *testing.T) { } peerClientConfig := &v1.Config{ + Version: migrations.CurrentVersion, Instance: defaultClientID, Repos: []*v1.Repo{ { @@ -231,10 +237,12 @@ func TestSyncConfigChange(t *testing.T) { // wait for the initial config to propagate tryExpectConfigFromHost(t, ctx, peerClient, peerClientConfig.Multihost.KnownHosts[0], &v1sync.RemoteConfig{ + Version: migrations.CurrentVersion, Repos: []*v1.Repo{ { Id: defaultRepoID, Guid: defaultRepoGUID, + Uri: "test-uri-should-sync", }, }, }) @@ -242,11 +250,15 @@ func TestSyncConfigChange(t *testing.T) { hostConfigChanged.Repos[0].Env = []string{"SOME_ENV=VALUE"} peerHost.configMgr.Update(hostConfigChanged) + tryConnect(t, ctx, peerClient, peerClientConfig.Multihost.KnownHosts[0]) + tryExpectConfigFromHost(t, ctx, peerClient, peerClientConfig.Multihost.KnownHosts[0], &v1sync.RemoteConfig{ + Version: migrations.CurrentVersion, Repos: []*v1.Repo{ { Id: defaultRepoID, Guid: defaultRepoGUID, + Uri: "test-uri-should-sync", Env: []string{"SOME_ENV=VALUE"}, }, }, @@ -254,7 +266,6 @@ func TestSyncConfigChange(t *testing.T) { } func TestSimpleOperationSync(t *testing.T) { - t.Skip("skipping syncapi test") testutil.InstallZapLogger(t) ctx, _ := context.WithTimeout(context.Background(), 5*time.Second) @@ -262,6 +273,7 @@ func TestSimpleOperationSync(t *testing.T) { peerClientAddr := testutil.AllocOpenBindAddr(t) peerHostConfig := &v1.Config{ + Version: migrations.CurrentVersion, Instance: defaultHostID, Repos: []*v1.Repo{ { @@ -282,6 +294,7 @@ func TestSimpleOperationSync(t *testing.T) { } peerClientConfig := &v1.Config{ + Version: migrations.CurrentVersion, Instance: defaultClientID, Repos: []*v1.Repo{ { @@ -378,7 +391,6 @@ func TestSimpleOperationSync(t *testing.T) { } func TestSyncMutations(t *testing.T) { - t.Skip("skipping syncapi tests") testutil.InstallZapLogger(t) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() @@ -387,6 +399,7 @@ func TestSyncMutations(t *testing.T) { peerClientAddr := testutil.AllocOpenBindAddr(t) peerHostConfig := &v1.Config{ + Version: migrations.CurrentVersion, Instance: defaultHostID, Repos: []*v1.Repo{ { @@ -407,6 +420,7 @@ func TestSyncMutations(t *testing.T) { } peerClientConfig := &v1.Config{ + Version: migrations.CurrentVersion, Instance: defaultClientID, Repos: []*v1.Repo{ { @@ -666,7 +680,7 @@ func tryConnect(t *testing.T, ctx context.Context, peer *peerUnderTest, hostPeer func runSyncAPIWithCtx(ctx context.Context, peer *peerUnderTest, bindAddr string) { mux := http.NewServeMux() syncHandler := NewBackrestSyncHandler(peer.manager) - mux.Handle(v1connect.NewBackrestSyncServiceHandler(syncHandler)) + mux.Handle(v1syncconnect.NewBackrestSyncServiceHandler(syncHandler)) server := &http.Server{ Addr: bindAddr, diff --git a/internal/api/syncapi/syncclient.go b/internal/api/syncapi/syncclient.go index c65b8725..c92f0165 100644 --- a/internal/api/syncapi/syncclient.go +++ b/internal/api/syncapi/syncclient.go @@ -407,9 +407,29 @@ func (c *syncSessionHandlerClient) HandleReceiveOperations(ctx context.Context, return NewSyncErrorProtocol(errors.New("client should not receive ReceiveOperations messages, this is a host-only message")) } +func (c *syncSessionHandlerClient) HandleReceiveResources(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionReceiveResources) error { + c.l.Debug("received resource list from server", + zap.Any("repos", item.GetRepos()), + zap.Any("plans", item.GetPlans())) + peerState := c.mgr.peerStateManager.GetPeerState(c.peer.Keyid).Clone() + if peerState == nil { + return NewSyncErrorInternal(fmt.Errorf("peer state for %q not found", c.peer.Keyid)) + } + repos := item.GetRepos() + plans := item.GetPlans() + for _, repo := range repos { + peerState.KnownRepos[repo.Id] = repo + } + for _, plan := range plans { + peerState.KnownPlans[plan.Id] = plan + } + c.mgr.peerStateManager.SetPeerState(c.peer.Keyid, peerState) + return nil +} + // Note unused: there isn't a situation where the host would send its config for information, the host will only call 'SetConfig' to update the config. func (c *syncSessionHandlerClient) HandleReceiveConfig(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionReceiveConfig) error { - c.l.Sugar().Debugf("received remote config update") + c.l.Sugar().Debugf("received remote config update", zap.Any("config", item.GetConfig())) peerState := c.mgr.peerStateManager.GetPeerState(c.peer.Keyid).Clone() if peerState == nil { return NewSyncErrorInternal(fmt.Errorf("peer state for %q not found", c.peer.Keyid))