fix: sync errors due to buffer reuse when streaming large batches of operation data

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)

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
 }
 

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
 }

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
 }

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
 		}

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)
 			}

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