lee/backrest
1
0
Fork 0
mirror of https://github.com/garethgeorge/backrest.git synced 2026-07-08 11:10:43 +00:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

start re-enabling tests for sync engine

Browse Source
This commit is contained in:
Gareth
2025-11-01 13:40:31 -07:00
parent 11b12c7c35
commit eae80020cd
4 changed files with 42 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/syncapi/cmdstreamutil.go
-3
View File
@@ -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:
internal/api/syncapi/handler.nogo
-288
View File
@@ -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 {
}
internal/api/syncapi/syncapi_test.nogo → internal/api/syncapi/syncapi_test.go
+21 -7
View File
@@ -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,
internal/api/syncapi/syncclient.go
+21 -1
View File
@@ -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))