diff --git a/internal/api/syncapi/syncapi_test.go b/internal/api/syncapi/syncapi_test.go index d0a4939e..d14112a1 100644 --- a/internal/api/syncapi/syncapi_test.go +++ b/internal/api/syncapi/syncapi_test.go @@ -74,7 +74,8 @@ var ( func TestConnectionSucceeds(t *testing.T) { testutil.InstallZapLogger(t) - ctx, _ := context.WithTimeout(context.Background(), 10*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() peerHostAddr := testutil.AllocOpenBindAddr(t) peerClientAddr := testutil.AllocOpenBindAddr(t) @@ -121,7 +122,8 @@ func TestConnectionSucceeds(t *testing.T) { func TestConnectionBadKeyRejected(t *testing.T) { testutil.InstallZapLogger(t) - ctx, _ := context.WithTimeout(context.Background(), 10*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() peerHostAddr := testutil.AllocOpenBindAddr(t) peerClientAddr := testutil.AllocOpenBindAddr(t) @@ -165,7 +167,8 @@ func TestConnectionBadKeyRejected(t *testing.T) { func TestSyncConfigChange(t *testing.T) { testutil.InstallZapLogger(t) - ctx, _ := context.WithTimeout(context.Background(), 10*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() peerHostAddr := testutil.AllocOpenBindAddr(t) peerClientAddr := testutil.AllocOpenBindAddr(t) @@ -270,7 +273,8 @@ func TestSyncConfigChange(t *testing.T) { func TestSimpleOperationSync(t *testing.T) { testutil.InstallZapLogger(t) - ctx, _ := context.WithTimeout(context.Background(), 10*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() peerHostAddr := testutil.AllocOpenBindAddr(t) peerClientAddr := testutil.AllocOpenBindAddr(t) diff --git a/internal/api/syncapi/syncclient.go b/internal/api/syncapi/syncclient.go index dbf050dd..8ef12d97 100644 --- a/internal/api/syncapi/syncclient.go +++ b/internal/api/syncapi/syncclient.go @@ -19,6 +19,7 @@ import ( "github.com/garethgeorge/backrest/gen/go/v1sync/v1syncconnect" "github.com/garethgeorge/backrest/internal/api/syncapi/permissions" "github.com/garethgeorge/backrest/internal/env" + "github.com/garethgeorge/backrest/internal/ioutil" "github.com/garethgeorge/backrest/internal/oplog" "go.uber.org/zap" "golang.org/x/net/http2" @@ -475,35 +476,30 @@ func (c *syncSessionHandlerClient) HandleOperationManifest(ctx context.Context, } func (c *syncSessionHandlerClient) HandleRequestOperationData(ctx context.Context, stream *bidiSyncCommandStream, item *v1sync.SyncStreamItem_SyncActionRequestOperationData) error { - var batch []*v1.Operation - send := func() { - if len(batch) == 0 { - return + for idBatch := range ioutil.Batchify(item.GetOpIds(), ioutil.DefaultBatchSize) { + ops := make([]*v1.Operation, 0, len(idBatch)) + for _, id := range idBatch { + op, err := c.oplog.Get(id) + if err != nil { + continue // may have been deleted between manifest and request + } + ops = append(ops, op) + } + if len(ops) == 0 { + continue } stream.Send(&v1sync.SyncStreamItem{ Action: &v1sync.SyncStreamItem_ReceiveOperations{ ReceiveOperations: &v1sync.SyncStreamItem_SyncActionReceiveOperations{ Event: &v1.OperationEvent{ Event: &v1.OperationEvent_UpdatedOperations{ - UpdatedOperations: &v1.OperationList{Operations: batch}, + UpdatedOperations: &v1.OperationList{Operations: ops}, }, }, }, }, }) - batch = batch[:0] } - for _, id := range item.GetOpIds() { - op, err := c.oplog.Get(id) - if err != nil { - continue // may have been deleted between manifest and request - } - batch = append(batch, op) - if len(batch) >= 256 { - send() - } - } - send() return nil } diff --git a/internal/ioutil/iobatching.go b/internal/ioutil/iobatching.go index df60fd00..7bc341ec 100644 --- a/internal/ioutil/iobatching.go +++ b/internal/ioutil/iobatching.go @@ -1,15 +1,16 @@ package ioutil +import "iter" + const DefaultBatchSize = 512 -func Batchify[T any](items []T, batchSize int) [][]T { - var batches [][]T - for i := 0; i < len(items); i += batchSize { - end := i + batchSize - if end > len(items) { - end = len(items) +func Batchify[T any](items []T, batchSize int) iter.Seq[[]T] { + return func(yield func([]T) bool) { + for i := 0; i < len(items); i += batchSize { + end := min(i+batchSize, len(items)) + if !yield(items[i:end]) { + return + } } - batches = append(batches, items[i:end]) } - return batches } diff --git a/internal/oplog/memstore/memstore.go b/internal/oplog/memstore/memstore.go index ca5e30a6..ef557c8b 100644 --- a/internal/oplog/memstore/memstore.go +++ b/internal/oplog/memstore/memstore.go @@ -151,7 +151,7 @@ func (m *MemStore) Add(op ...*v1.Operation) error { } for _, o := range op { - m.operations[o.Id] = o + m.operations[o.Id] = proto.Clone(o).(*v1.Operation) } return nil } @@ -163,7 +163,7 @@ func (m *MemStore) Get(opID int64) (*v1.Operation, error) { if !ok { return nil, oplog.ErrNotExist } - return op, nil + return proto.Clone(op).(*v1.Operation), nil } func (m *MemStore) Delete(opID ...int64) ([]*v1.Operation, error) { @@ -172,7 +172,7 @@ func (m *MemStore) Delete(opID ...int64) ([]*v1.Operation, error) { ops := make([]*v1.Operation, 0, len(opID)) for _, id := range opID { if op, ok := m.operations[id]; ok { - ops = append(ops, op) + ops = append(ops, proto.Clone(op).(*v1.Operation)) } delete(m.operations, id) } @@ -198,7 +198,7 @@ func (m *MemStore) Set(opts oplog.SetOptions, op ...*v1.Operation) error { if err := protoutil.ValidateOperation(o); err != nil { return err } - m.operations[o.Id] = o + m.operations[o.Id] = proto.Clone(o).(*v1.Operation) } else { if o.Modno == 0 { m.nextModno++ @@ -212,7 +212,7 @@ func (m *MemStore) Set(opts oplog.SetOptions, op ...*v1.Operation) error { if _, ok := m.operations[o.Id]; !ok { return oplog.ErrNotExist } - m.operations[o.Id] = o + m.operations[o.Id] = proto.Clone(o).(*v1.Operation) } } return nil @@ -230,7 +230,7 @@ func (m *MemStore) Update(op ...*v1.Operation) error { if _, ok := m.operations[o.Id]; !ok { return oplog.ErrNotExist } - m.operations[o.Id] = o + m.operations[o.Id] = proto.Clone(o).(*v1.Operation) } return nil } diff --git a/internal/oplog/migrations.go b/internal/oplog/migrations.go index 6c7ee505..e14a66d2 100644 --- a/internal/oplog/migrations.go +++ b/internal/oplog/migrations.go @@ -123,7 +123,7 @@ func migration003DeduplicateIndexedSnapshots(oplog *OpLog) error { return nil } - for _, batch := range ioutil.Batchify(deleteIDs, ioutil.DefaultBatchSize) { + for batch := range ioutil.Batchify(deleteIDs, ioutil.DefaultBatchSize) { if _, err := oplog.store.Delete(batch...); err != nil { return err } diff --git a/internal/oplog/oplog.go b/internal/oplog/oplog.go index f04e2bef..b091028a 100644 --- a/internal/oplog/oplog.go +++ b/internal/oplog/oplog.go @@ -7,6 +7,7 @@ import ( "sync" v1 "github.com/garethgeorge/backrest/gen/go/v1" + "google.golang.org/protobuf/proto" ) type OperationEvent int @@ -55,9 +56,17 @@ func (o *OpLog) curSubscribers() []subAndQuery { } func (o *OpLog) notify(ops []*v1.Operation, event OperationEvent) { - for _, sub := range o.curSubscribers() { - notifyOps := make([]*v1.Operation, 0, len(ops)) - for _, op := range ops { + subs := o.curSubscribers() + if len(subs) == 0 { + return + } + cloned := make([]*v1.Operation, len(ops)) + for i, op := range ops { + cloned[i] = proto.Clone(op).(*v1.Operation) + } + for _, sub := range subs { + notifyOps := make([]*v1.Operation, 0, len(cloned)) + for _, op := range cloned { if sub.q.Match(op) { notifyOps = append(notifyOps, op) } diff --git a/internal/oplog/sqlitestore/sqlitestore.go b/internal/oplog/sqlitestore/sqlitestore.go index 7248601c..f70a3a17 100644 --- a/internal/oplog/sqlitestore/sqlitestore.go +++ b/internal/oplog/sqlitestore/sqlitestore.go @@ -610,7 +610,7 @@ func (m *SqliteStore) Delete(opID ...int64) ([]*v1.Operation, error) { defer tx.Rollback() ops := make([]*v1.Operation, 0, len(opID)) - for _, batch := range ioutil.Batchify(opID, ioutil.DefaultBatchSize) { + for batch := range ioutil.Batchify(opID, ioutil.DefaultBatchSize) { batchOps, err := m.deleteHelper(tx, batch...) if err != nil { return nil, err