feeds: fix sequence feed resolution, drop timebase heuristic (#1259)

Co-authored-by: Metacertain <metacertain@gmail.com>

pkg/api/feed_test.go

@@ -271,3 +271,11 @@ type id struct {
 func (i *id) MarshalBinary() ([]byte, error) {
 	return []byte("accd"), nil
 }
+
+func (i *id) String() string {
+	return "44237"
+}
+
+func (*id) Next(last int64, at uint64) feeds.Index {
+	return &id{}
+}

pkg/feeds/epochs/epoch.go

@@ -8,6 +8,7 @@ package epochs
 
 import (
 	"encoding/binary"
+	"fmt"
 
 	"github.com/ethersphere/bee/pkg/crypto"
 	"github.com/ethersphere/bee/pkg/feeds"
@@ -19,12 +20,17 @@ const (
 
 var _ feeds.Index = (*epoch)(nil)
 
-// epoch is referencing a slot in the epoch grid
+// epoch is referencing a slot in the epoch grid and represents an update
+// it  implements the feeds.Index interface
 type epoch struct {
 	start uint64
 	level uint8
 }
 
+func (e *epoch) String() string {
+	return fmt.Sprintf("%d/%d", e.start, e.level)
+}
+
 // MarshalBinary implements the BinaryMarshaler interface
 func (e *epoch) MarshalBinary() ([]byte, error) {
 	epochBytes := make([]byte, 8)
@@ -32,6 +38,21 @@ func (e *epoch) MarshalBinary() ([]byte, error) {
 	return crypto.LegacyKeccak256(append(epochBytes, e.level))
 }
 
+func next(e feeds.Index, last int64, at uint64) feeds.Index {
+	if e == nil {
+		return &epoch{0, maxLevel}
+	}
+	return e.Next(last, at)
+}
+
+// Next implements feeds.Index advancement
+func (e *epoch) Next(last int64, at uint64) feeds.Index {
+	if e.start+e.length() > at {
+		return e.childAt(at)
+	}
+	return lca(int64(at), last).childAt(at)
+}
+
 // lca calculates the lowest common ancestor epoch given two unix times
 func lca(at, after int64) *epoch {
 	if after == 0 {
@@ -48,16 +69,6 @@ func lca(at, after int64) *epoch {
 	return &epoch{start, level}
 }
 
-func next(e *epoch, last int64, at uint64) *epoch {
-	if e == nil {
-		return &epoch{0, maxLevel}
-	}
-	if e.start+e.length() > at {
-		return e.childAt(at)
-	}
-	return lca(int64(at), last).childAt(at)
-}
-
 // parent returns the ancestor of an epoch
 // the call is unsafe in that it must not be called on a toplevel epoch
 func (e *epoch) parent() *epoch {

pkg/feeds/epochs/lookup_benchmark_test.go

@@ -6,44 +6,21 @@ package epochs_test
 
 import (
 	"context"
-	"errors"
 	"fmt"
-	"math/rand"
 	"testing"
-	"time"
 
 	"github.com/ethersphere/bee/pkg/crypto"
 	"github.com/ethersphere/bee/pkg/feeds"
 	"github.com/ethersphere/bee/pkg/feeds/epochs"
-	"github.com/ethersphere/bee/pkg/storage"
+	feedstesting "github.com/ethersphere/bee/pkg/feeds/testing"
 	"github.com/ethersphere/bee/pkg/storage/mock"
-	"github.com/ethersphere/bee/pkg/swarm"
 )
 
-type timeout struct {
-	storage.Storer
-}
-
-var searchTimeout = 30 * time.Millisecond
-
-// Get overrides the mock storer and introduces latency
-func (t *timeout) Get(ctx context.Context, mode storage.ModeGet, addr swarm.Address) (swarm.Chunk, error) {
-	ch, err := t.Storer.Get(ctx, mode, addr)
-	if err != nil {
-		if errors.Is(err, storage.ErrNotFound) {
-			time.Sleep(searchTimeout)
-		}
-		return ch, err
-	}
-	time.Sleep(time.Duration(rand.Intn(10)) * time.Millisecond)
-	return ch, nil
-}
-
 func BenchmarkFinder(b *testing.B) {
 	for _, i := range []int{0, 8, 30} {
 		for _, prefill := range []int64{1, 50} {
 			after := int64(50)
-			storer := &timeout{mock.NewStorer()}
+			storer := &feedstesting.Timeout{Storer: mock.NewStorer()}
 			topicStr := "testtopic"
 			topic, err := crypto.LegacyKeccak256([]byte(topicStr))
 			if err != nil {

pkg/feeds/epochs/lookup_test.go

@@ -20,8 +20,13 @@ func TestFinder(t *testing.T) {
 		t.Run("basic", func(t *testing.T) {
 			feedstesting.TestFinderBasic(t, finderf, updaterf)
 		})
+		i := int64(0)
+		nextf := func() (bool, int64) {
+			defer func() { i++ }()
+			return i < 50, i
+		}
 		t.Run("fixed", func(t *testing.T) {
-			feedstesting.TestFinderFixIntervals(t, finderf, updaterf)
+			feedstesting.TestFinderFixIntervals(t, nextf, finderf, updaterf)
 		})
 		t.Run("random", func(t *testing.T) {
 			feedstesting.TestFinderRandomIntervals(t, finderf, updaterf)

pkg/feeds/epochs/updater.go

@@ -19,7 +19,7 @@ var _ feeds.Updater = (*updater)(nil)
 type updater struct {
 	*feeds.Putter
 	last  int64
-	epoch *epoch
+	epoch feeds.Index
 }
 
 // NewUpdater constructs a feed updater

pkg/feeds/feed.go

@@ -28,6 +28,7 @@ type Factory interface {
 	NewLookup(Type, *Feed) (Lookup, error)
 }
 
+// Type enumerates the time-based feed types
 type Type int
 
 const (
@@ -46,6 +47,7 @@ func (t Type) String() string {
 	}
 }
 
+// FromString constructs the type from a string
 func (t *Type) FromString(s string) error {
 	switch s = strings.ToLower(s); s {
 	case "sequence":
@@ -66,7 +68,7 @@ type id struct {
 var _ encoding.BinaryMarshaler = (*id)(nil)
 
 func (i *id) MarshalBinary() ([]byte, error) {
-	return crypto.LegacyKeccak256(append(i.topic, i.index...))
+	return crypto.LegacyKeccak256(append(append([]byte{}, i.topic...), i.index...))
 }
 
 // Feed is representing an epoch based feed
@@ -84,6 +86,8 @@ func New(topic []byte, owner common.Address) *Feed {
 // Index is the interface for feed implementations.
 type Index interface {
 	encoding.BinaryMarshaler
+	Next(last int64, at uint64) Index
+	fmt.Stringer
 }
 
 // Update represents an update instance of a feed, i.e., pairing of a Feed with an Epoch
@@ -97,6 +101,7 @@ func (f *Feed) Update(index Index) *Update {
 	return &Update{f, index}
 }
 
+// NewUpdate creates an update from an index, timestamp, payload and signature
 func NewUpdate(f *Feed, idx Index, timestamp int64, payload []byte, sig []byte) (swarm.Chunk, error) {
 	id, err := f.Update(idx).Id()
 	if err != nil {
@@ -119,12 +124,18 @@ func NewUpdate(f *Feed, idx Index, timestamp int64, payload []byte, sig []byte)
 
 // Id calculates the identifier if a  feed update to be used in single owner chunks
 func (u *Update) Id() ([]byte, error) {
-	index, err := u.index.MarshalBinary()
+	return Id(u.Topic, u.index)
+}
+
+// Id calculates the feed id from a topic and an index
+func Id(topic []byte, index Index) ([]byte, error) {
+	indexBytes, err := index.MarshalBinary()
 	if err != nil {
 		return nil, err
 	}
-	i := &id{u.Topic, index}
+	i := &id{topic, indexBytes}
 	return i.MarshalBinary()
+
 }
 
 // Address calculates the soc address of a feed update

pkg/feeds/sequence/lookup_benchmark_test.go

@@ -6,42 +6,19 @@ package sequence_test
 
 import (
 	"context"
-	"errors"
 	"fmt"
-	"math/rand"
 	"testing"
-	"time"
 
 	"github.com/ethersphere/bee/pkg/crypto"
 	"github.com/ethersphere/bee/pkg/feeds"
 	"github.com/ethersphere/bee/pkg/feeds/sequence"
-	"github.com/ethersphere/bee/pkg/storage"
+	feedstesting "github.com/ethersphere/bee/pkg/feeds/testing"
 	"github.com/ethersphere/bee/pkg/storage/mock"
-	"github.com/ethersphere/bee/pkg/swarm"
 )
 
-type timeout struct {
-	storage.Storer
-}
-
-var searchTimeout = 30 * time.Millisecond
-
-// Get overrides the mock storer and introduces latency
-func (t *timeout) Get(ctx context.Context, mode storage.ModeGet, addr swarm.Address) (swarm.Chunk, error) {
-	ch, err := t.Storer.Get(ctx, mode, addr)
-	if err != nil {
-		if errors.Is(err, storage.ErrNotFound) {
-			time.Sleep(searchTimeout)
-		}
-		return ch, err
-	}
-	time.Sleep(time.Duration(rand.Intn(10)) * time.Millisecond)
-	return ch, nil
-}
-
 func BenchmarkFinder(b *testing.B) {
 	for _, prefill := range []int64{1, 100, 1000, 5000} {
-		storer := &timeout{mock.NewStorer()}
+		storer := &feedstesting.Timeout{Storer: mock.NewStorer()}
 		topicStr := "testtopic"
 		topic, err := crypto.LegacyKeccak256([]byte(topicStr))
 		if err != nil {

pkg/feeds/sequence/lookup_test.go

@@ -19,8 +19,13 @@ func TestFinder(t *testing.T) {
 		t.Run("basic", func(t *testing.T) {
 			feedstesting.TestFinderBasic(t, finderf, updaterf)
 		})
+		i := 0
+		nextf := func() (bool, int64) {
+			i++
+			return i == 40, int64(i)
+		}
 		t.Run("fixed", func(t *testing.T) {
-			feedstesting.TestFinderFixIntervals(t, finderf, updaterf)
+			feedstesting.TestFinderFixIntervals(t, nextf, finderf, updaterf)
 		})
 		t.Run("random", func(t *testing.T) {
 			feedstesting.TestFinderRandomIntervals(t, finderf, updaterf)

pkg/feeds/sequence/sequence.go

@@ -14,6 +14,7 @@ import (
 	"context"
 	"encoding/binary"
 	"errors"
+	"fmt"
 
 	"github.com/ethersphere/bee/pkg/crypto"
 	"github.com/ethersphere/bee/pkg/feeds"
@@ -21,28 +22,42 @@ import (
 	"github.com/ethersphere/bee/pkg/swarm"
 )
 
+// DefaultLevels is the number of concurrent lookaheads
+// 8 spans 2^8 updates
+const DefaultLevels = 8
+
 var _ feeds.Index = (*index)(nil)
 var _ feeds.Lookup = (*finder)(nil)
 var _ feeds.Lookup = (*asyncFinder)(nil)
 var _ feeds.Updater = (*updater)(nil)
 
+// index just wraps a uint64. implements the feeds.Index interface
 type index struct {
 	index uint64
 }
 
+func (i *index) String() string {
+	return fmt.Sprintf("%d", i.index)
+}
+
 func (i *index) MarshalBinary() ([]byte, error) {
 	indexBytes := make([]byte, 8)
 	binary.BigEndian.PutUint64(indexBytes, i.index)
 	return indexBytes, nil
 }
 
+// Next requires
+func (i *index) Next(last int64, at uint64) feeds.Index {
+	return &index{i.index + 1}
+}
+
 // finder encapsulates a chunk store getter and a feed and provides
-//  non-concurrent lookup methods
+// non-concurrent lookup
 type finder struct {
 	getter *feeds.Getter
 }
 
-// NewFinder constructs an Finder
+// NewFinder constructs an finder (feeds.Lookup interface)
 func NewFinder(getter storage.Getter, feed *feeds.Feed) feeds.Lookup {
 	return &finder{feeds.NewGetter(getter, feed)}
 }
@@ -56,21 +71,25 @@ func (f *finder) At(ctx context.Context, at, after int64) (ch swarm.Chunk, curre
 			if !errors.Is(err, storage.ErrNotFound) {
 				return nil, nil, nil, err
 			}
-			return ch, &index{i - 1}, &index{i}, nil
+			if i > 0 {
+				current = &index{i - 1}
+			}
+			return ch, current, &index{i}, nil
 		}
 		ts, err := feeds.UpdatedAt(u)
 		if err != nil {
 			return nil, nil, nil, err
 		}
+		// if timestamp is later than the `at` target datetime, then return previous chunk  and index
 		if ts > uint64(at) {
-			return ch, &index{i}, nil, nil
+			return ch, &index{i - 1}, &index{i}, nil
 		}
 		ch = u
 	}
 }
 
 // asyncFinder encapsulates a chunk store getter and a feed and provides
-//  non-concurrent lookup methods
+//  non-concurrent lookup
 type asyncFinder struct {
 	getter *feeds.Getter
 }
@@ -80,132 +99,157 @@ func NewAsyncFinder(getter storage.Getter, feed *feeds.Feed) feeds.Lookup {
 	return &asyncFinder{feeds.NewGetter(getter, feed)}
 }
 
-type path struct {
-	latest    result
-	base      uint64
-	level     int
-	cancel    chan struct{}
-	cancelled bool
+// interval represents a batch of concurrent retreieve requests
+// that probe the interval (base,b+2^level) at offsets 2^k-1 for k=1,...,max
+// recording  the level of the latest found update chunk and the earliest not found update
+// the actual latest update is guessed to be within a subinterval
+type interval struct {
+	base     uint64  // beginning of the interval, guaranteed to have an  update
+	level    int     // maximum level to check
+	found    *result // the result with the latest chunk found
+	notFound int     // the earliest level where no update is found
 }
 
-func (p *path) close() {
-	if !p.cancelled {
-		close(p.cancel)
-		p.cancelled = true
+// when a subinterval is identified to contain the latest update
+// next returns an interval matching it
+func (i *interval) next() *interval {
+	found := i.found.level
+	i.found.level = 0
+	return &interval{
+		base:     i.found.index, // set base to index of latest chunk found
+		level:    found,         // set max level to the latest update level
+		notFound: found,         // set notFound to the latest update level
+		found:    i.found,       // inherit latest found  result
 	}
 }
 
-func newPath(base uint64) *path {
-	return &path{base: base, cancel: make(chan struct{})}
+func (i *interval) retry() *interval {
+	r := i.next()
+	r.level = i.level    // reset to max
+	r.notFound = i.level //  reset to max
+	return r
+}
+
+func newInterval(base uint64) *interval {
+	return &interval{base: base, level: DefaultLevels, notFound: DefaultLevels}
 }
 
+// results capture a chunk lookup on a interval
 type result struct {
-	chunk swarm.Chunk
-	path  *path
-	level int
-	seq   uint64
-	diff  int64
+	chunk    swarm.Chunk // the chunk found
+	interval *interval   // the interval it belongs to
+	level    int         // the level within the interval
+	index    uint64      // the actual seqeuence index of the update
 }
 
 // At looks up the version valid at time `at`
 // after is a unix time hint of the latest known update
 func (f *asyncFinder) At(ctx context.Context, at, after int64) (ch swarm.Chunk, cur, next feeds.Index, err error) {
-	ch, diff, err := f.get(ctx, at, 0)
+	// first lookup update at the 0 index
+	ch, err = f.get(ctx, at, 0)
 	if err != nil {
 		return nil, nil, nil, err
 	}
 	if ch == nil {
-		return nil, nil, nil, nil
-	}
-	if diff == 0 {
-		return ch, &index{0}, &index{1}, nil
-	}
-	c := make(chan result)
-	p := newPath(0)
-	p.latest.chunk = ch
-	for p.level = 1; diff>>p.level > 0; p.level++ {
+		return nil, nil, &index{0}, nil
 	}
+	// if chunk exists construct an initial interval with base=0
+	c := make(chan *result)
+	i := newInterval(0)
+	i.found = &result{ch, nil, 0, 0}
+
 	quit := make(chan struct{})
 	defer close(quit)
-	go f.at(ctx, at, p, c, quit)
+
+	// launch concurrent request at  doubling intervals
+	go f.at(ctx, at, 0, i, c, quit)
 	for r := range c {
-		p = r.path
+		// collect the results into the interval
+		i = r.interval
 		if r.chunk == nil {
-			if r.level == 0 {
-				return p.latest.chunk, &index{p.latest.seq}, &index{p.latest.seq + 1}, nil
-			}
-			if p.level < r.level {
+			if i.notFound < r.level {
 				continue
 			}
-			p.level = r.level - 1
+			i.notFound = r.level - 1
 		} else {
-			if r.diff == 0 {
-				return r.chunk, &index{r.seq}, &index{r.seq + 1}, nil
-			}
-			if p.latest.level > r.level {
+			if i.found.level > r.level {
 				continue
 			}
-			p.close()
-			p.latest = r
+			// if a chunk is found on the max level, and this is already a subinterval
+			// then found.index+1 is already known to be not found
+			if i.level == r.level && r.level < DefaultLevels {
+				return r.chunk, &index{r.index}, &index{r.index + 1}, nil
+			}
+			i.found = r
 		}
-		// below applies even  if  p.latest==maxLevel
-		if p.latest.level == p.level {
-			if p.level == 0 {
-				return p.latest.chunk, &index{p.latest.seq}, &index{p.latest.seq + 1}, nil
+		// below applies even  if  i.latest==ceilingLevel in which case we just continue with
+		// DefaultLevel lookaheads
+		if i.found.level == i.notFound {
+			if i.found.level == 0 {
+				return i.found.chunk, &index{i.found.index}, &index{i.found.index + 1}, nil
 			}
-			p.close()
-			np := newPath(p.latest.seq)
-			np.level = p.level
-			np.latest.chunk = p.latest.chunk
-			go f.at(ctx, at, np, c, quit)
+			go f.at(ctx, at, 0, i.next(), c, quit)
+		}
+		// inconsistent feed, retry
+		if i.notFound < i.found.level {
+			go f.at(ctx, at, i.found.level, i.retry(), c, quit)
 		}
 	}
 	return nil, nil, nil, nil
 }
 
-func (f *asyncFinder) at(ctx context.Context, at int64, p *path, c chan<- result, quit <-chan struct{}) {
-	for i := p.level; i > 0; i-- {
+// at launches concurrent lookups at exponential intervals after th c starting from further
+func (f *asyncFinder) at(ctx context.Context, at int64, min int, i *interval, c chan<- *result, quit <-chan struct{}) {
+	stop := make(chan struct{}, 1)
+	for l := i.level; l > min; l-- {
 		select {
-		case <-p.cancel:
+		case <-stop: // if a chunk is found
 			return
-		case <-quit:
+		case <-quit: // if the parent process quit
 			return
 		default:
 		}
-		go func(i int) {
-			seq := p.base + (1 << i) - 1
-			ch, diff, err := f.get(ctx, at, seq)
+		go func(l int) {
+			index := i.base + (1 << l) - 1
+			ch, err := f.get(ctx, at, index)
 			if err != nil {
 				return
 			}
+			// if a chunk is found, stop the iterationq
+			if ch != nil {
+				select {
+				case stop <- struct{}{}:
+				default:
+				}
+			}
 			select {
-			case c <- result{ch, p, i, seq, diff}:
+			case c <- &result{ch, i, l, index}:
 			case <-quit:
 			}
-		}(i)
+		}(l)
 	}
 }
 
-func (f *asyncFinder) get(ctx context.Context, at int64, seq uint64) (swarm.Chunk, int64, error) {
-	u, err := f.getter.Get(ctx, &index{seq})
+// get performs a lookup of an update chunk, returns nil (not error) if not found
+func (f *asyncFinder) get(ctx context.Context, at int64, idx uint64) (swarm.Chunk, error) {
+	u, err := f.getter.Get(ctx, &index{idx})
 	if err != nil {
 		if !errors.Is(err, storage.ErrNotFound) {
-			return nil, 0, err
+			return nil, err
 		}
 		// if 'not-found' error, then just silence and return nil chunk
-		return nil, 0, nil
+		return nil, nil
 	}
 	ts, err := feeds.UpdatedAt(u)
 	if err != nil {
-		return nil, 0, err
+		return nil, err
 	}
-	diff := at - int64(ts)
 	// this means the update timestamp is later than the pivot time we are looking for
 	// handled as if the update was missing but with no uncertainty due to timeout
-	if diff < 0 {
-		return nil, 0, nil
+	if at < int64(ts) {
+		return nil, nil
 	}
-	return u, diff, nil
+	return u, nil
 }
 
 // updater encapsulates a feeds putter to generate successive updates for epoch based feeds

pkg/feeds/testing/lookup.go

@@ -8,6 +8,8 @@ package testing
 import (
 	"bytes"
 	"context"
+	"encoding/binary"
+	"errors"
 	"fmt"
 	"math/rand"
 	"testing"
@@ -17,10 +19,30 @@ import (
 	"github.com/ethersphere/bee/pkg/feeds"
 	"github.com/ethersphere/bee/pkg/storage"
 	"github.com/ethersphere/bee/pkg/storage/mock"
+	"github.com/ethersphere/bee/pkg/swarm"
 )
 
+type Timeout struct {
+	storage.Storer
+}
+
+var searchTimeout = 30 * time.Millisecond
+
+// Get overrides the mock storer and introduces latency
+func (t *Timeout) Get(ctx context.Context, mode storage.ModeGet, addr swarm.Address) (swarm.Chunk, error) {
+	ch, err := t.Storer.Get(ctx, mode, addr)
+	if err != nil {
+		if errors.Is(err, storage.ErrNotFound) {
+			time.Sleep(searchTimeout)
+		}
+		return ch, err
+	}
+	time.Sleep(time.Duration(rand.Intn(10)) * time.Millisecond)
+	return ch, nil
+}
+
 func TestFinderBasic(t *testing.T, finderf func(storage.Getter, *feeds.Feed) feeds.Lookup, updaterf func(putter storage.Putter, signer crypto.Signer, topic []byte) (feeds.Updater, error)) {
-	storer := mock.NewStorer()
+	storer := &Timeout{mock.NewStorer()}
 	topicStr := "testtopic"
 	topic, err := crypto.LegacyKeccak256([]byte(topicStr))
 	if err != nil {
@@ -73,128 +95,114 @@ func TestFinderBasic(t *testing.T, finderf func(storage.Getter, *feeds.Feed) fee
 	})
 }
 
-func TestFinderFixIntervals(t *testing.T, finderf func(storage.Getter, *feeds.Feed) feeds.Lookup, updaterf func(putter storage.Putter, signer crypto.Signer, topic []byte) (feeds.Updater, error)) {
-	for _, tc := range []struct {
-		count  int64
-		step   int64
-		offset int64
-	}{
-		{50, 1, 0},
-		{50, 1, 10000},
-		{50, 100, 0},
-		{50, 100, 100000},
-	} {
-		t.Run(fmt.Sprintf("count=%d,step=%d,offset=%d", tc.count, tc.step, tc.offset), func(t *testing.T) {
-			storer := mock.NewStorer()
-			topicStr := "testtopic"
-			topic, err := crypto.LegacyKeccak256([]byte(topicStr))
+func TestFinderFixIntervals(t *testing.T, nextf func() (bool, int64), finderf func(storage.Getter, *feeds.Feed) feeds.Lookup, updaterf func(putter storage.Putter, signer crypto.Signer, topic []byte) (feeds.Updater, error)) {
+	var stop bool
+	for j := 10; !stop; j += 10 {
+		t.Run(fmt.Sprintf("custom intervals up to %d", j), func(t *testing.T) {
+			var i int64
+			var n int
+			f := func() (bool, int64) {
+				n++
+				stop, i = nextf()
+				return n == j || stop, i
+			}
+			TestFinderIntervals(t, f, finderf, updaterf)
+		})
+	}
+}
+
+func TestFinderIntervals(t *testing.T, nextf func() (bool, int64), finderf func(storage.Getter, *feeds.Feed) feeds.Lookup, updaterf func(putter storage.Putter, signer crypto.Signer, topic []byte) (feeds.Updater, error)) {
+
+	storer := &Timeout{mock.NewStorer()}
+	topicStr := "testtopic"
+	topic, err := crypto.LegacyKeccak256([]byte(topicStr))
+	if err != nil {
+		t.Fatal(err)
+	}
+	pk, _ := crypto.GenerateSecp256k1Key()
+	signer := crypto.NewDefaultSigner(pk)
+
+	updater, err := updaterf(storer, signer, topic)
+	if err != nil {
+		t.Fatal(err)
+	}
+	finder := finderf(storer, updater.Feed())
+
+	ctx := context.Background()
+	var ats []int64
+	for stop, at := nextf(); !stop; stop, at = nextf() {
+		ats = append(ats, at)
+		payload := make([]byte, 8)
+		binary.BigEndian.PutUint64(payload, uint64(at))
+		err = updater.Update(ctx, at, payload)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	for j := 0; j < len(ats)-1; j++ {
+		at := ats[j]
+		diff := ats[j+1] - at
+		for now := at; now < ats[j+1]; now += int64(rand.Intn(int(diff)) + 1) {
+			after := int64(0)
+			ch, current, next, err := finder.At(ctx, now, after)
 			if err != nil {
 				t.Fatal(err)
 			}
-			pk, _ := crypto.GenerateSecp256k1Key()
-			signer := crypto.NewDefaultSigner(pk)
-
-			updater, err := updaterf(storer, signer, topic)
+			if ch == nil {
+				t.Fatalf("expected to find update, got none")
+			}
+			ts, payload, err := feeds.FromChunk(ch)
 			if err != nil {
 				t.Fatal(err)
 			}
-			ctx := context.Background()
+			content := binary.BigEndian.Uint64(payload)
+			if content != uint64(at) {
+				t.Fatalf("payload mismatch: expected %v, got %v", at, content)
+			}
 
-			payload := []byte("payload")
-			for at := tc.offset; at < tc.offset+tc.count*tc.step; at += tc.step {
-				err = updater.Update(ctx, at, payload)
+			if ts != uint64(at) {
+				t.Fatalf("timestamp mismatch: expected %v, got %v", at, ts)
+			}
+
+			if current != nil {
+				expectedId := ch.Data()[:32]
+				id, err := feeds.Id(topic, current)
 				if err != nil {
 					t.Fatal(err)
 				}
+				if !bytes.Equal(id, expectedId) {
+					t.Fatalf("current mismatch: expected %x, got %x", expectedId, id)
+				}
 			}
-			finder := finderf(storer, updater.Feed())
-			for at := tc.offset; at < tc.offset+tc.count*tc.step; at += tc.step {
-				for after := tc.offset; after < at; after += tc.step {
-					step := int64(1)
-					if tc.step > 1 {
-						step = tc.step / 4
-					}
-					for now := at; now < at+tc.step; now += step {
-						ch, _, _, err := finder.At(ctx, now, after)
-						if err != nil {
-							t.Fatal(err)
-						}
-						if ch == nil {
-							t.Fatalf("expected to find update, got none")
-						}
-						exp := payload
-						ts, payload, err := feeds.FromChunk(ch)
-						if err != nil {
-							t.Fatal(err)
-						}
-						if !bytes.Equal(payload, exp) {
-							t.Fatalf("payload mismatch: expected %x, got %x", exp, payload)
-						}
-						if ts != uint64(at) {
-							t.Fatalf("timestamp mismatch: expected %v, got %v", at, ts)
-						}
-					}
+			if next != nil {
+				expectedNext := current.Next(at, uint64(now))
+				expectedIdx, err := expectedNext.MarshalBinary()
+				if err != nil {
+					t.Fatal(err)
+				}
+				idx, err := next.MarshalBinary()
+				if err != nil {
+					t.Fatal(err)
+				}
+				if !bytes.Equal(idx, expectedIdx) {
+					t.Fatalf("next mismatch: expected %x, got %x", expectedIdx, idx)
 				}
 			}
-		})
+		}
 	}
 }
 
 func TestFinderRandomIntervals(t *testing.T, finderf func(storage.Getter, *feeds.Feed) feeds.Lookup, updaterf func(putter storage.Putter, signer crypto.Signer, topic []byte) (feeds.Updater, error)) {
-	for i := 0; i < 5; i++ {
-		t.Run(fmt.Sprintf("random intervals %d", i), func(t *testing.T) {
-			storer := mock.NewStorer()
-			topicStr := "testtopic"
-			topic, err := crypto.LegacyKeccak256([]byte(topicStr))
-			if err != nil {
-				t.Fatal(err)
-			}
-			pk, _ := crypto.GenerateSecp256k1Key()
-			signer := crypto.NewDefaultSigner(pk)
-
-			updater, err := updaterf(storer, signer, topic)
-			if err != nil {
-				t.Fatal(err)
-			}
-			ctx := context.Background()
-
-			payload := []byte("payload")
-			var at int64
-			ats := make([]int64, 100)
-			for j := 0; j < 50; j++ {
-				ats[j] = at
-				at += int64(rand.Intn(1<<10) + 1)
-				err = updater.Update(ctx, ats[j], payload)
-				if err != nil {
-					t.Fatal(err)
-				}
-			}
-			finder := finderf(storer, updater.Feed())
-			for j := 1; j < 49; j++ {
-				diff := ats[j+1] - ats[j]
-				for at := ats[j]; at < ats[j+1]; at += int64(rand.Intn(int(diff)) + 1) {
-					for after := int64(0); after < at; after += int64(rand.Intn(int(at))) {
-						ch, _, _, err := finder.At(ctx, at, after)
-						if err != nil {
-							t.Fatal(err)
-						}
-						if ch == nil {
-							t.Fatalf("expected to find update, got none")
-						}
-						exp := payload
-						ts, payload, err := feeds.FromChunk(ch)
-						if err != nil {
-							t.Fatal(err)
-						}
-						if !bytes.Equal(payload, exp) {
-							t.Fatalf("payload mismatch: expected %x, got %x", exp, payload)
-						}
-						if ts != uint64(ats[j]) {
-							t.Fatalf("timestamp mismatch: expected %v, got %v", ats[j], ts)
-						}
-					}
-				}
+	for j := 0; j < 3; j++ {
+		t.Run(fmt.Sprintf("random intervals %d", j), func(t *testing.T) {
+			var i int64
+			var n int
+			nextf := func() (bool, int64) {
+				i += int64(rand.Intn(1<<10) + 1)
+				n++
+				return n == 40, i
 			}
+			TestFinderIntervals(t, nextf, finderf, updaterf)
 		})
 	}
 }