watch ranges over transaction limit

melgenek · melgenek · commit b97989a22d85 · 2025-08-19T01:39:53.000+02:00
diff --git a/README.md b/README.md
@@ -44,7 +44,6 @@ The implementation is in the `pkg/drivers/fdb` directory.
 ## TODO
 
 Here is a list of implementation details that need to be completed before starting scale testing of this implementation.
-- [ ] Watch operation has to account for long-running or large transactions. Make sure that in such case there are multiple consecutive FDB transactions. 
 - [ ] put ttl events into a separate slice
 - [ ] limit records to 2MiB
 
@@ -89,5 +88,6 @@ Here is a list of implementation details that need to be completed before starti
 - https://forums.foundationdb.org/t/relax-consistency-guarantees/1560/19
 - https://forums.foundationdb.org/t/generating-sortable-unique-id-primary-key-across-the-cluster-in-an-entity-relationship-model/3789/2
 - https://forums.foundationdb.org/t/foundationdb-read-performance/729/4
+- https://forums.foundationdb.org/t/ranges-without-explicit-end-go/773/13
 - 
 
diff --git a/pkg/drivers/fdb/fdb_compact.go b/pkg/drivers/fdb/fdb_compact.go
@@ -28,20 +28,23 @@ func (c *compactProcessor) next(tr *fdb.Transaction, it *fdb.RangeIterator) (fdb
 	if err != nil {
 		return nil, false, err
 	}
+	if rev == nil {
+		return nil, false, nil
+	}
 
 	lastRecord, err := c.f.getLast(tr, record.Key)
 	if err != nil {
 		return nil, false, err
 	}
 
-	if lastRecord.Key.Rev != rev || record.IsDelete {
-		c.f.byKeyAndRevision.Delete(tr, &KeyAndRevision{Key: record.Key, Rev: rev})
-		if err := c.f.byRevision.Delete(tr, rev); err != nil {
+	if lastRecord.Key.Rev != *rev || record.IsDelete {
+		c.f.byKeyAndRevision.Delete(tr, &KeyAndRevision{Key: record.Key, Rev: *rev})
+		if err := c.f.byRevision.Delete(tr, *rev); err != nil {
 			return nil, false, err
 		}
 	}
-	c.batchCompacted = rev
-	return c.f.byRevision.GetSubspace().Pack(tuple.Tuple{rev}), true, nil
+	c.batchCompacted = *rev
+	return c.f.byRevision.GetSubspace().Pack(tuple.Tuple{*rev}), true, nil
 }
 
 func (c *compactProcessor) endBatch(tr *fdb.Transaction, isLast bool) error {
diff --git a/pkg/drivers/fdb/fdb_read.go b/pkg/drivers/fdb/fdb_read.go
@@ -145,7 +145,7 @@ func (c *listCollector) next(tr *fdb.Transaction, record *ByKeyAndRevisionRecord
 	}
 	c.batchIterators = append(c.batchIterators, recordIt)
 
-	if len(c.batchIterators) > 2 {
+	if len(c.batchIterators) >= 1 {
 		if err := c.fetchIterators(); err != nil {
 			return nil, false, err
 		}
@@ -154,7 +154,7 @@ func (c *listCollector) next(tr *fdb.Transaction, record *ByKeyAndRevisionRecord
 }
 
 func (c *listCollector) needMore() bool {
-	return c.limit == 0 || int64(len(c.batchIterators)+len(c.records)) < c.limit
+	return c.limit == 0 || int64(len(c.batchIterators)+len(c.batchRecords)+len(c.records)) < c.limit
 }
 
 func (c *listCollector) endBatch(*fdb.Transaction, bool) error {
@@ -166,18 +166,18 @@ func (c *listCollector) fetchIterators() error {
 		rev, record, err := c.f.byRevision.GetFromIterator(it)
 		if err != nil {
 			return err
+		} else if rev == nil {
+			return fmt.Errorf("no records in by rev iterator")
 		} else {
-			c.batchRecords = append(c.batchRecords, &RevRecord{Rev: rev, Record: record})
+			c.batchRecords = append(c.batchRecords, &RevRecord{Rev: *rev, Record: record})
 		}
 	}
 	c.batchIterators = c.batchIterators[len(c.batchIterators):]
 	return nil
 }
 
 func (c *listCollector) postBatch() {
-	for _, record := range c.batchRecords {
-		c.records = append(c.records, record)
-	}
+	c.records = append(c.records, c.batchRecords...)
 }
 
 func (c *listCollector) String() string {
@@ -213,6 +213,10 @@ func (c *recordCollector) next(tr *fdb.Transaction, it *fdb.RangeIterator) (fdb.
 	if err != nil {
 		return nil, false, err
 	}
+	if nextKeyAndRevRecord == nil {
+		return nil, false, nil
+	}
+
 	if c.batchCurrentRecord != nil && c.batchCurrentRecord.Key.Key != nextKeyAndRevRecord.Key.Key {
 		if !c.batchCurrentRecord.Value.IsDelete {
 			if _, _, err := c.inner.next(tr, c.batchCurrentRecord); err != nil {
diff --git a/pkg/drivers/fdb/fdb_subspace_by_key_revision.go b/pkg/drivers/fdb/fdb_subspace_by_key_revision.go
@@ -61,6 +61,9 @@ func (s *ByKeyAndRevisionSubspace) Delete(tr *fdb.Transaction, key *KeyAndRevisi
 }
 
 func (s *ByKeyAndRevisionSubspace) GetFromIterator(it *fdb.RangeIterator) (*ByKeyAndRevisionRecord, error) {
+	if !it.Advance() {
+		return nil, nil
+	}
 	kv, err := it.Get()
 	if err != nil {
 		return nil, err
diff --git a/pkg/drivers/fdb/fdb_subspace_by_revision.go b/pkg/drivers/fdb/fdb_subspace_by_revision.go
@@ -111,28 +111,28 @@ func (s *ByRevisionSubspace) GetIterator(tr *fdb.Transaction, rev tuple.Versions
 	return it, nil
 }
 
-func (s *ByRevisionSubspace) GetFromIterator(it *fdb.RangeIterator) (tuple.Versionstamp, *Record, error) {
+func (s *ByRevisionSubspace) GetFromIterator(it *fdb.RangeIterator) (*tuple.Versionstamp, *Record, error) {
 	if !it.Advance() {
-		return dummyVersionstamp, nil, nil
+		return nil, nil, nil
 	}
 	kv, err := it.Get()
 	if err != nil {
-		return dummyVersionstamp, nil, err
+		return nil, nil, err
 	}
 	rev, record, err := s.ParseKV(kv)
 	if err != nil {
-		return dummyVersionstamp, nil, err
+		return nil, nil, err
 	}
 	if record.ValueSize > int64(len(record.Value)) {
 		buf := make([]byte, record.ValueSize)
 		copy(buf, record.Value)
 		offset := len(record.Value)
 		if err := s.getBlob(it, buf, offset); err != nil {
-			return dummyVersionstamp, nil, err
+			return nil, nil, err
 		}
 		record.Value = buf
 	}
-	return rev, record, nil
+	return &rev, record, nil
 }
 
 func (s *ByRevisionSubspace) getBlob(it *fdb.RangeIterator, buf []byte, offset int) error {
diff --git a/pkg/drivers/fdb/fdb_test.go b/pkg/drivers/fdb/fdb_test.go
@@ -3,11 +3,12 @@ package fdb
 import (
 	"bytes"
 	"context"
-	"crypto/rand"
 	"fmt"
 	"github.com/stretchr/testify/assert"
 	"golang.org/x/sync/errgroup"
+	"math/rand"
 	"slices"
+	"strings"
 	"testing"
 	"time"
 
@@ -18,7 +19,7 @@ import (
 )
 
 func TestFDB(t *testing.T) {
-	forceRetryTransaction = func(i int) bool { return i < 2 }
+	forceRetryTransaction = func(i int) bool { return false }
 
 	logrus.SetLevel(logrus.InfoLevel)
 	n := 4
@@ -247,9 +248,19 @@ func TestFDBLargeRecords(t *testing.T) {
 	err := f.Start(ctx)
 	require.NoError(t, err)
 
+	watchLarge := f.Watch(ctx, "/large/", 0)
+
 	recordCount := 300
 	records := createLargeRecords(t, f, ctx, recordCount)
 
+	for i := 0; i < recordCount; {
+		batch := <-watchLarge.Events
+		for _, watchedEvent := range batch {
+			i++
+			require.Equal(t, records[watchedEvent.KV.Key], watchedEvent.KV.Value, "watched value for key '%s' does not match", watchedEvent.KV.Key)
+		}
+	}
+
 	// Get record
 	_, kv, err := f.Get(ctx, "/large/key42", "", 0, 0)
 	require.NoError(t, err)
@@ -442,20 +453,29 @@ func createLargeRecords(t *testing.T, f server.Backend, ctx context.Context, rec
 	records := make(map[string][]byte, recordCount)
 	for i := 0; i < recordCount; i++ {
 		key := fmt.Sprintf("/large/key%d", i)
-		value := make([]byte, recordSize)
-		_, err := rand.Read(value)
-		require.NoError(t, err)
-		records[key] = value
+		value := randomString(recordSize)
+		records[key] = []byte(value)
 		v := func(key string, value []byte) (e error) {
 			_, err := f.Create(ctx, key, value, 0)
 			return err
 		}
-		g.Go(func() error { return v(key, value) })
+		g.Go(func() error { return v(key, []byte(value)) })
 	}
 	require.NoError(t, g.Wait())
 	return records
 }
 
+const letters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
+
+func randomString(n int) string {
+	var sb strings.Builder
+	sb.Grow(n)
+	for i := 0; i < n; i++ {
+		sb.WriteByte(letters[rand.Intn(len(letters))])
+	}
+	return sb.String()
+}
+
 func orEmpty(result []*server.KeyValue) []*server.KeyValue {
 	if result == nil {
 		result = make([]*server.KeyValue, 0)
diff --git a/pkg/drivers/fdb/fdb_ttl.go b/pkg/drivers/fdb/fdb_ttl.go
@@ -118,7 +118,7 @@ func (f *FDB) ttlEvents(ctx context.Context) chan *server.Event {
 		defer close(result)
 
 		collector := newListCollector(f, 1000)
-		rev, err := f.listWithCollector("ttl1", "/", "/", 0, collector)
+		rev, err := f.listWithCollector("ttl1", "/", "", 0, collector)
 		revRecords := collector.records
 		for len(revRecords) > 0 {
 			if err != nil {
diff --git a/pkg/drivers/fdb/fdb_util.go b/pkg/drivers/fdb/fdb_util.go
@@ -15,6 +15,7 @@ const (
 	logConflictingKeys = false
 
 	splitRangeAfterDuration  = 1 * time.Second
+	transactionTimeout       = 10 * time.Second
 	transactionMaxRetryCount = 1000
 )
 
@@ -30,33 +31,38 @@ type Processor[T any] interface {
 type batchResult struct {
 	lastReadKey        fdb.KeyConvertible
 	collectorNeedsMore bool
-	iteratorHasMore    bool
 }
 
 func processRange(db fdb.Database, selector fdb.SelectorRange, collector Processor[*fdb.RangeIterator]) error {
 	beginSelector := selector.Begin
 
 	for i := 0; ; i++ {
-		res, err := collectBatch(db, fdb.SelectorRange{Begin: beginSelector, End: selector.End}, collector)
+		res, err := processBatch(db, fdb.SelectorRange{Begin: beginSelector, End: selector.End}, collector)
 		if err != nil {
 			return err
 		}
 		if !res.collectorNeedsMore {
 			break
 		}
-		if !res.iteratorHasMore {
-			break
-		}
 		beginSelector = fdb.FirstGreaterThan(res.lastReadKey)
 	}
 
 	return nil
 }
 
-func collectBatch(db fdb.Database, selector fdb.SelectorRange, collector Processor[*fdb.RangeIterator]) (batchResult, error) {
+func processBatch(db fdb.Database, selector fdb.SelectorRange, collector Processor[*fdb.RangeIterator]) (batchResult, error) {
+	before := time.Now()
+	defer func() {
+		dur := time.Since(before)
+
+		if dur > 2*splitRangeAfterDuration {
+			logrus.Warnf("BATCH %s => duration=%s", selector, dur)
+		}
+	}()
+
 	res, err := transact(db, batchResult{}, func(tr fdb.Transaction) (batchResult, error) {
-		res := batchResult{collectorNeedsMore: true, iteratorHasMore: true}
-		if err := tr.Options().SetTimeout(2 * splitRangeAfterDuration.Milliseconds()); err != nil {
+		res := batchResult{collectorNeedsMore: true}
+		if err := tr.Options().SetTimeout(transactionTimeout.Milliseconds()); err != nil {
 			return res, fmt.Errorf("failed to set timeout limit: %w", err)
 		}
 
@@ -67,26 +73,22 @@ func collectBatch(db fdb.Database, selector fdb.SelectorRange, collector Process
 
 		var firstKey fdb.KeyConvertible
 		collector.startBatch()
-		for i := 0; res.collectorNeedsMore && res.iteratorHasMore; i++ {
-			if !it.Advance() {
-				res.iteratorHasMore = false
-				break
-			}
+		for i := 0; res.collectorNeedsMore; i++ {
 			if lastKey, collectorNeedsMore, err := collector.next(&tr, it); err != nil {
 				return res, err
 			} else {
 				if firstKey == nil {
 					firstKey = lastKey
 				}
 				res.lastReadKey = lastKey
-				res.collectorNeedsMore = collectorNeedsMore
+				res.collectorNeedsMore = collectorNeedsMore && lastKey != nil
 			}
 			if time.Since(start) > splitRangeAfterDuration {
 				break
 			}
 		}
 
-		if err := collector.endBatch(&tr, !res.iteratorHasMore); err != nil {
+		if err := collector.endBatch(&tr, !res.collectorNeedsMore); err != nil {
 			return res, err
 		}
 
diff --git a/pkg/drivers/fdb/fdb_watch.go b/pkg/drivers/fdb/fdb_watch.go

Original file line number	Diff line number	Diff line change
`@@ -28,20 +28,23 @@ func (c compactProcessor) next(tr fdb.Transaction, it *fdb.RangeIterator) (fdb`
`28`	`28`	`if err != nil {`
`29`	`29`	`return nil, false, err`
`30`	`30`	`}`
	`31`	`+ if rev == nil {`
	`32`	`+ return nil, false, nil`
	`33`	`+ }`
`31`	`34`
`32`	`35`	`lastRecord, err := c.f.getLast(tr, record.Key)`
`33`	`36`	`if err != nil {`
`34`	`37`	`return nil, false, err`
`35`	`38`	`}`
`36`	`39`
`37`		`- if lastRecord.Key.Rev != rev \|\| record.IsDelete {`
`38`		`- c.f.byKeyAndRevision.Delete(tr, &KeyAndRevision{Key: record.Key, Rev: rev})`
`39`		`- if err := c.f.byRevision.Delete(tr, rev); err != nil {`
	`40`	`+ if lastRecord.Key.Rev != *rev \|\| record.IsDelete {`
	`41`	`+ c.f.byKeyAndRevision.Delete(tr, &KeyAndRevision{Key: record.Key, Rev: *rev})`
	`42`	`+ if err := c.f.byRevision.Delete(tr, *rev); err != nil {`
`40`	`43`	`return nil, false, err`
`41`	`44`	`}`
`42`	`45`	`}`
`43`		`- c.batchCompacted = rev`
`44`		`- return c.f.byRevision.GetSubspace().Pack(tuple.Tuple{rev}), true, nil`
	`46`	`+ c.batchCompacted = *rev`
	`47`	`+ return c.f.byRevision.GetSubspace().Pack(tuple.Tuple{*rev}), true, nil`
`45`	`48`	`}`
`46`	`49`
`47`	`50`	`func (c compactProcessor) endBatch(tr fdb.Transaction, isLast bool) error {`
Original file line number	Diff line number	Diff line change
`@@ -145,7 +145,7 @@ func (c listCollector) next(tr fdb.Transaction, record *ByKeyAndRevisionRecord`
`145`	`145`	`}`
`146`	`146`	`c.batchIterators = append(c.batchIterators, recordIt)`
`147`	`147`
`148`		`- if len(c.batchIterators) > 2 {`
	`148`	`+ if len(c.batchIterators) >= 1 {`
`149`	`149`	`if err := c.fetchIterators(); err != nil {`
`150`	`150`	`return nil, false, err`
`151`	`151`	`}`
`@@ -154,7 +154,7 @@ func (c listCollector) next(tr fdb.Transaction, record *ByKeyAndRevisionRecord`
`154`	`154`	`}`
`155`	`155`
`156`	`156`	`func (c *listCollector) needMore() bool {`
`157`		`- return c.limit == 0 \|\| int64(len(c.batchIterators)+len(c.records)) < c.limit`
	`157`	`+ return c.limit == 0 \|\| int64(len(c.batchIterators)+len(c.batchRecords)+len(c.records)) < c.limit`
`158`	`158`	`}`
`159`	`159`
`160`	`160`	`func (c listCollector) endBatch(fdb.Transaction, bool) error {`
`@@ -166,18 +166,18 @@ func (c *listCollector) fetchIterators() error {`
`166`	`166`	`rev, record, err := c.f.byRevision.GetFromIterator(it)`
`167`	`167`	`if err != nil {`
`168`	`168`	`return err`
	`169`	`+ } else if rev == nil {`
	`170`	`+ return fmt.Errorf("no records in by rev iterator")`
`169`	`171`	`} else {`
`170`		`- c.batchRecords = append(c.batchRecords, &RevRecord{Rev: rev, Record: record})`
	`172`	`+ c.batchRecords = append(c.batchRecords, &RevRecord{Rev: *rev, Record: record})`
`171`	`173`	`}`
`172`	`174`	`}`
`173`	`175`	`c.batchIterators = c.batchIterators[len(c.batchIterators):]`
`174`	`176`	`return nil`
`175`	`177`	`}`
`176`	`178`
`177`	`179`	`func (c *listCollector) postBatch() {`
`178`		`- for _, record := range c.batchRecords {`
`179`		`- c.records = append(c.records, record)`
`180`		`- }`
	`180`	`+ c.records = append(c.records, c.batchRecords...)`
`181`	`181`	`}`
`182`	`182`
`183`	`183`	`func (c *listCollector) String() string {`
`@@ -213,6 +213,10 @@ func (c recordCollector) next(tr fdb.Transaction, it *fdb.RangeIterator) (fdb.`
`213`	`213`	`if err != nil {`
`214`	`214`	`return nil, false, err`
`215`	`215`	`}`
	`216`	`+ if nextKeyAndRevRecord == nil {`
	`217`	`+ return nil, false, nil`
	`218`	`+ }`
	`219`	`+`
`216`	`220`	`if c.batchCurrentRecord != nil && c.batchCurrentRecord.Key.Key != nextKeyAndRevRecord.Key.Key {`
`217`	`221`	`if !c.batchCurrentRecord.Value.IsDelete {`
`218`	`222`	`if _, _, err := c.inner.next(tr, c.batchCurrentRecord); err != nil {`
Original file line number	Diff line number	Diff line change
`@@ -61,6 +61,9 @@ func (s ByKeyAndRevisionSubspace) Delete(tr fdb.Transaction, key *KeyAndRevisi`
`61`	`61`	`}`
`62`	`62`
`63`	`63`	`func (s ByKeyAndRevisionSubspace) GetFromIterator(it fdb.RangeIterator) (*ByKeyAndRevisionRecord, error) {`
	`64`	`+ if !it.Advance() {`
	`65`	`+ return nil, nil`
	`66`	`+ }`
`64`	`67`	`kv, err := it.Get()`
`65`	`68`	`if err != nil {`
`66`	`69`	`return nil, err`
Original file line number	Diff line number	Diff line change
`@@ -111,28 +111,28 @@ func (s ByRevisionSubspace) GetIterator(tr fdb.Transaction, rev tuple.Versions`
`111`	`111`	`return it, nil`
`112`	`112`	`}`
`113`	`113`
`114`		`-func (s ByRevisionSubspace) GetFromIterator(it fdb.RangeIterator) (tuple.Versionstamp, *Record, error) {`
	`114`	`+func (s ByRevisionSubspace) GetFromIterator(it fdb.RangeIterator) (tuple.Versionstamp, Record, error) {`
`115`	`115`	`if !it.Advance() {`
`116`		`- return dummyVersionstamp, nil, nil`
	`116`	`+ return nil, nil, nil`
`117`	`117`	`}`
`118`	`118`	`kv, err := it.Get()`
`119`	`119`	`if err != nil {`
`120`		`- return dummyVersionstamp, nil, err`
	`120`	`+ return nil, nil, err`
`121`	`121`	`}`
`122`	`122`	`rev, record, err := s.ParseKV(kv)`
`123`	`123`	`if err != nil {`
`124`		`- return dummyVersionstamp, nil, err`
	`124`	`+ return nil, nil, err`
`125`	`125`	`}`
`126`	`126`	`if record.ValueSize > int64(len(record.Value)) {`
`127`	`127`	`buf := make([]byte, record.ValueSize)`
`128`	`128`	`copy(buf, record.Value)`
`129`	`129`	`offset := len(record.Value)`
`130`	`130`	`if err := s.getBlob(it, buf, offset); err != nil {`
`131`		`- return dummyVersionstamp, nil, err`
	`131`	`+ return nil, nil, err`
`132`	`132`	`}`
`133`	`133`	`record.Value = buf`
`134`	`134`	`}`
`135`		`- return rev, record, nil`
	`135`	`+ return &rev, record, nil`
`136`	`136`	`}`
`137`	`137`
`138`	`138`	`func (s ByRevisionSubspace) getBlob(it fdb.RangeIterator, buf []byte, offset int) error {`