sync2: get rid of contexts and improve logging and sequence handling

Author: ivan4th

sync2/rangesync/dumbset.go

@@ -1,7 +1,6 @@
 package rangesync
 
 import (
-	"context"
 	"crypto/md5"
 	"errors"
 	"slices"
@@ -95,7 +94,7 @@ type dumbSet struct {
 
 var _ OrderedSet = &dumbSet{}
 
-func (ds *dumbSet) Receive(ctx context.Context, id types.KeyBytes) error {
+func (ds *dumbSet) Receive(id types.KeyBytes) error {
 	if len(ds.keys) == 0 {
 		ds.keys = []types.KeyBytes{id}
 		return nil
@@ -124,22 +123,25 @@ func (ds *dumbSet) Receive(ctx context.Context, id types.KeyBytes) error {
 	return nil
 }
 
-func (ds *dumbSet) seq(n int) types.Seq {
+func (ds *dumbSet) seq(n int) types.SeqResult {
 	if n < 0 || n > len(ds.keys) {
 		panic("bad index")
 	}
-	return types.Seq(func(yield func(types.KeyBytes, error) bool) {
-		n := n // make the sequence reusable
-		for {
-			if !yield(ds.keys[n], nil) {
-				break
+	return types.SeqResult{
+		Seq: types.Seq(func(yield func(types.KeyBytes) bool) {
+			n := n // make the sequence reusable
+			for {
+				if !yield(ds.keys[n]) {
+					break
+				}
+				n = (n + 1) % len(ds.keys)
 			}
-			n = (n + 1) % len(ds.keys)
-		}
-	})
+		}),
+		Error: types.NoSeqError,
+	}
 }
 
-func (ds *dumbSet) seqFor(s types.KeyBytes) types.Seq {
+func (ds *dumbSet) seqFor(s types.KeyBytes) types.SeqResult {
 	n := slices.IndexFunc(ds.keys, func(k types.KeyBytes) bool {
 		return k.Compare(s) == 0
 	})
@@ -150,7 +152,6 @@ func (ds *dumbSet) seqFor(s types.KeyBytes) types.Seq {
 }
 
 func (ds *dumbSet) getRangeInfo(
-	_ context.Context,
 	x, y types.KeyBytes,
 	count int,
 ) (r RangeInfo, end types.KeyBytes, err error) {
@@ -179,35 +180,29 @@ func (ds *dumbSet) getRangeInfo(
 			panic("empty start/end from naiveRange")
 		}
 		r.Items = ds.seqFor(start)
+	} else {
+		r.Items = types.EmptySeqResult()
 	}
 	return r, end, nil
 }
 
-func (ds *dumbSet) GetRangeInfo(
-	ctx context.Context,
-	x, y types.KeyBytes,
-	count int,
-) (RangeInfo, error) {
-	ri, _, err := ds.getRangeInfo(ctx, x, y, count)
+func (ds *dumbSet) GetRangeInfo(x, y types.KeyBytes, count int) (RangeInfo, error) {
+	ri, _, err := ds.getRangeInfo(x, y, count)
 	return ri, err
 }
 
-func (ds *dumbSet) SplitRange(
-	ctx context.Context,
-	x, y types.KeyBytes,
-	count int,
-) (SplitInfo, error) {
+func (ds *dumbSet) SplitRange(x, y types.KeyBytes, count int) (SplitInfo, error) {
 	if count <= 0 {
 		panic("BUG: bad split count")
 	}
-	part0, middle, err := ds.getRangeInfo(ctx, x, y, count)
+	part0, middle, err := ds.getRangeInfo(x, y, count)
 	if err != nil {
 		return SplitInfo{}, err
 	}
 	if part0.Count == 0 {
 		return SplitInfo{}, errors.New("can't split empty range")
 	}
-	part1, err := ds.GetRangeInfo(ctx, middle, y, -1)
+	part1, err := ds.GetRangeInfo(middle, y, -1)
 	if err != nil {
 		return SplitInfo{}, err
 	}
@@ -217,23 +212,23 @@ func (ds *dumbSet) SplitRange(
 	}, nil
 }
 
-func (ds *dumbSet) Empty(ctx context.Context) (bool, error) {
+func (ds *dumbSet) Empty() (bool, error) {
 	return len(ds.keys) == 0, nil
 }
 
-func (ds *dumbSet) Items(ctx context.Context) (types.Seq, error) {
+func (ds *dumbSet) Items() types.SeqResult {
 	if len(ds.keys) == 0 {
-		return types.EmptySeq(), nil
+		return types.EmptySeqResult()
 	}
-	return ds.seq(0), nil
+	return ds.seq(0)
 }
 
-func (ds *dumbSet) Copy() OrderedSet {
+func (ds *dumbSet) Copy(syncScope bool) OrderedSet {
 	return &dumbSet{keys: slices.Clone(ds.keys)}
 }
 
-func (ds *dumbSet) Recent(ctx context.Context, since time.Time) (types.Seq, int, error) {
-	return nil, 0, nil
+func (ds *dumbSet) Recent(since time.Time) (types.SeqResult, int) {
+	return types.EmptySeqResult(), 0
 }
 
 var hashPool = &sync.Pool{
@@ -267,7 +262,7 @@ type deferredAddSet struct {
 	added map[string]struct{}
 }
 
-func (das *deferredAddSet) Receive(ctx context.Context, id types.KeyBytes) error {
+func (das *deferredAddSet) Receive(id types.KeyBytes) error {
 	if das.added == nil {
 		das.added = make(map[string]struct{})
 	}
@@ -276,9 +271,8 @@ func (das *deferredAddSet) Receive(ctx context.Context, id types.KeyBytes) error
 }
 
 func (das *deferredAddSet) addAll() error {
-	ctx := context.Background()
 	for k := range das.added {
-		if err := das.OrderedSet.Receive(ctx, types.KeyBytes(k)); err != nil {
+		if err := das.OrderedSet.Receive(types.KeyBytes(k)); err != nil {
 			return err
 		}
 	}

sync2/rangesync/helpers.go

@@ -1,28 +1,20 @@
 package rangesync
 
 import (
-	"context"
-
 	"github.com/spacemeshos/go-spacemesh/sync2/types"
 )
 
 // CollectSetItems returns the list of items in the given set.
-func CollectSetItems(ctx context.Context, os OrderedSet) (r []types.KeyBytes, err error) {
-	items, err := os.Items(ctx)
-	if err != nil {
-		return nil, err
-	}
+func CollectSetItems(os OrderedSet) (r []types.KeyBytes, err error) {
+	items := os.Items()
 	var first types.KeyBytes
-	for v, err := range items {
-		if err != nil {
-			return nil, err
-		}
+	for v := range items.Seq {
 		if first == nil {
 			first = v
 		} else if v.Compare(first) == 0 {
 			break
 		}
 		r = append(r, v)
 	}
-	return r, nil
+	return r, items.Error()
 }

sync2/rangesync/interface.go

@@ -17,7 +17,7 @@ type RangeInfo struct {
 	// Number of items in the interval
 	Count int
 	// Items is the sequence of set elements in the interval.
-	Items types.Seq
+	Items types.SeqResult
 }
 
 // SplitInfo contains information about range split in two.
@@ -33,7 +33,7 @@ type OrderedSet interface {
 	// Receive handles a new key received from the peer.
 	// It may or may not add it to the set immediately; this doesn't affect set
 	// reconciliation operation.
-	Receive(ctx context.Context, k types.KeyBytes) error
+	Receive(k types.KeyBytes) error
 	// GetRangeInfo returns RangeInfo for the item range in the ordered set,
 	// bounded by [x, y).
 	// x == y indicates the whole set.
@@ -44,20 +44,24 @@ type OrderedSet interface {
 	// is returned for the corresponding subrange of the requested range.
 	// If both x and y are nil, the information for the entire set is returned.
 	// If any of x or y is nil, the other one must be nil as well.
-	GetRangeInfo(ctx context.Context, x, y types.KeyBytes, count int) (RangeInfo, error)
+	GetRangeInfo(x, y types.KeyBytes, count int) (RangeInfo, error)
 	// SplitRange splits the range roughly after the specified count of items,
 	// returning RangeInfo for the first half and the second half of the range.
-	SplitRange(ctx context.Context, x, y types.KeyBytes, count int) (SplitInfo, error)
+	SplitRange(x, y types.KeyBytes, count int) (SplitInfo, error)
 	// Items returns the sequence of items in the set.
-	Items(ctx context.Context) (types.Seq, error)
+	Items() types.SeqResult
 	// Empty returns true if the set is empty.
-	Empty(ctx context.Context) (bool, error)
+	Empty() (bool, error)
 	// Copy makes a shallow copy of the OrderedSet.
-	Copy() OrderedSet
+	// syncScope argument is a hint that can be used to optimize resource usage.
+	// If syncScope is true, then the copy is intended to be used for the duration of
+	// a synchronization run.
+	// If syncScope if false, then the lifetime of the copy is not clearly defined.
+	Copy(syncScope bool) OrderedSet
 	// Recent returns an Iterator that yields the items added since the specified
 	// timestamp. Some OrderedSet implementations may not have Recent implemented, in
-	// which case it should return an error.
-	Recent(ctx context.Context, since time.Time) (types.Seq, int, error)
+	// which case it should return an empty sequence.
+	Recent(since time.Time) (types.SeqResult, int)
 }
 
 type Requester interface {

sync2/rangesync/log.go

@@ -153,9 +153,9 @@ func (s sender) sendSample(
 	x, y types.KeyBytes,
 	fp types.Fingerprint,
 	count, sampleSize int,
-	seq types.Seq,
+	sr types.SeqResult,
 ) error {
-	items, err := Sample(seq, count, sampleSize)
+	items, err := Sample(sr, count, sampleSize)
 	if err != nil {
 		return err
 	}

sync2/rangesync/message.go

@@ -44,22 +44,19 @@ func MinhashSampleItemFromKeyBytes(h types.KeyBytes) MinhashSampleItem {
 
 // Sample retrieves min(count, sampleSize) items friom the ordered sequence, extracting
 // MinhashSampleItem from each value.
-func Sample(seq types.Seq, count, sampleSize int) ([]MinhashSampleItem, error) {
+func Sample(sr types.SeqResult, count, sampleSize int) ([]MinhashSampleItem, error) {
 	sampleSize = min(count, sampleSize)
 	if sampleSize == 0 {
 		return nil, nil
 	}
 	items := make([]MinhashSampleItem, 0, sampleSize)
-	for k, err := range seq {
-		if err != nil {
-			return nil, err
-		}
+	for k := range sr.Seq {
 		items = append(items, MinhashSampleItemFromKeyBytes(k))
 		if len(items) == sampleSize {
 			break
 		}
 	}
-	return items, nil
+	return items, sr.Error()
 }
 
 // CalcSim estimates the Jaccard similarity coefficient between two sets based on the

sync2/rangesync/minhash.go

@@ -11,19 +11,25 @@ import (
 
 var errBadItem = errors.New("bad item")
 
-func mkFakeSeq(items []string) types.Seq {
-	return func(yield func(types.KeyBytes, error) bool) {
-		for {
-			for _, item := range items {
-				if item == "ERROR" {
-					yield(nil, errBadItem)
-					return
-				}
-				if !yield(types.KeyBytes(item), nil) {
-					return
+func mkFakeSeq(items []string) types.SeqResult {
+	var err error
+	return types.SeqResult{
+		Seq: func(yield func(types.KeyBytes) bool) {
+			for {
+				for _, item := range items {
+					if item == "ERROR" {
+						err = errBadItem
+						return
+					}
+					if !yield(types.KeyBytes(item)) {
+						return
+					}
 				}
 			}
-		}
+		},
+		Error: func() error {
+			return err
+		},
 	}
 }