Repository: textileio/go-threads
Pull request #477 (2020-12-24 11:50:12)

Thread edge proposal
Hi folks, here is another improvement for GetRecords.

This time we propose a new concept: thread edge. Basically it's just a 64-bit deterministic hash of all heads in a thread. Using the edge values we can easily compare two states of the same thread and decide if they are equal without fetching all thread's information from the logstore. Such optimization at first should improve request processing time on a fast path for the frequent GetRecords requests.
As a follow-up: introducing thread edges enables implementing fast checks for thread changes without getting and sending thread information on both  communicating nodes (protocol extension in progress for now).



Patch:
diff --git a/core/logstore/logstore.go b/core/logstore/logstore.go
index 55f3a79..5e236bb 100644
--- a/core/logstore/logstore.go
+++ b/core/logstore/logstore.go
@@ -203,6 +203,9 @@ type HeadBook interface {
 	// ClearHeads deletes the head entry for a log.
 	ClearHeads(thread.ID, peer.ID) error
 
+	// HeadsEdge returns deterministic hash of all heads of a given thread.
+	HeadsEdge(t thread.ID) (uint64, error)
+
 	// DumpHeads packs entire headbook into the tree.
 	DumpHeads() (DumpHeadBook, error)
 
diff --git a/db/index.go b/db/index.go
index 3c23cab..6adb8e8 100644
--- a/db/index.go
+++ b/db/index.go
@@ -290,8 +290,10 @@ func newIterator(txn dse.TxnExt, baseKey ds.Key, q *Query) *iterator {
 	dsq := dse.QueryExt{
 		Query: query.Query{
 			Prefix: prefix.String(),
-			Limit:  q.Limit,
-			Offset: q.Skip,
+			// Pull out Skip and Limit here because we won't know ahead of time
+			// (due to readFilters) how many to skip/limit
+			// Limit:  q.Limit,
+			// Offset: q.Skip,
 		},
 	}
 	if q.Sort.FieldPath == idFieldName {
diff --git a/db/listeners.go b/db/listeners.go
index cba1b15..0be1ddd 100644
--- a/db/listeners.go
+++ b/db/listeners.go
@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"sync"
 
-	"github.com/ipfs/go-ipld-format"
+	format "github.com/ipfs/go-ipld-format"
 	"github.com/textileio/go-threads/core/app"
 	core "github.com/textileio/go-threads/core/db"
 	"github.com/textileio/go-threads/core/thread"
@@ -107,8 +107,6 @@ func (scn *stateChangedNotifee) addListener(sl *listener) {
 }
 
 func (scn *stateChangedNotifee) remove(sl *listener) bool {
-	scn.lock.Lock()
-	defer scn.lock.Unlock()
 	for i := range scn.listeners {
 		if scn.listeners[i] == sl {
 			scn.listeners[i] = scn.listeners[len(scn.listeners)-1]
diff --git a/db/query.go b/db/query.go
index 28f030e..66e1ea9 100644
--- a/db/query.go
+++ b/db/query.go
@@ -314,6 +314,9 @@ func (t *Txn) Find(q *Query) ([][]byte, error) {
 		return nil, err
 	}
 	var values []MarshaledResult
+	// Use count to track real count of returned values taking into account
+	// read filter and any indexes etc in the query
+	var count = 0
 	for {
 		res, ok := iter.NextSync()
 		if !ok {
@@ -324,7 +327,14 @@ func (t *Txn) Find(q *Query) ([][]byte, error) {
 			return nil, err
 		}
 		if res.Value != nil {
-			values = append(values, res)
+			// Only count valid values that aren't filtered by the read filter
+			count++
+			if count > q.Skip {
+				values = append(values, res)
+			}
+		}
+		if len(values) == q.Limit {
+			break
 		}
 	}
 
diff --git a/logstore/lstoreds/headbook.go b/logstore/lstoreds/headbook.go
index 3d5016d..a6e5991 100644
--- a/logstore/lstoreds/headbook.go
+++ b/logstore/lstoreds/headbook.go
@@ -1,6 +1,7 @@
 package lstoreds
 
 import (
+	"encoding/binary"
 	"fmt"
 
 	"github.com/gogo/protobuf/proto"
@@ -11,17 +12,23 @@ import (
 	core "github.com/textileio/go-threads/core/logstore"
 	"github.com/textileio/go-threads/core/thread"
 	pb "github.com/textileio/go-threads/net/pb"
+	"github.com/textileio/go-threads/util"
 )
 
 type dsHeadBook struct {
 	ds ds.TxnDatastore
 }
 
-// Heads are stored in db key pattern:
-// /thread/heads/<base32 thread id no padding>/<base32 peer id no padding>
 var (
-	hbBase               = ds.NewKey("/thread/heads")
-	_      core.HeadBook = (*dsHeadBook)(nil)
+	// Heads are stored in db key pattern:
+	// /thread/heads/<base32 thread id no padding>/<base32 peer id no padding>
+	hbBase = ds.NewKey("/thread/heads")
+
+	// Heads edges are stored in db key pattern:
+	// /thread/heads:edge/<base32 thread id no padding>>
+	hbEdge = ds.NewKey("/thread/heads:edge")
+
+	_ core.HeadBook = (*dsHeadBook)(nil)
 )
 
 // NewHeadBook returns a new HeadBook backed by a datastore.
@@ -69,12 +76,12 @@ func (hb *dsHeadBook) AddHeads(t thread.ID, p peer.ID, heads []cid.Cid) error {
 			hr.Heads = append(hr.Heads, entry)
 		}
 	}
-	data, err := proto.Marshal(&hr)
-	if err != nil {
+	if data, err := proto.Marshal(&hr); err != nil {
 		return fmt.Errorf("error when marshaling headbookrecord proto for %v: %w", key, err)
-	}
-	if err = txn.Put(key, data); err != nil {
+	} else if err = txn.Put(key, data); err != nil {
 		return fmt.Errorf("error when saving new head record in datastore for %v: %v", key, err)
+	} else if err := hb.invalidateEdge(txn, t); err != nil {
+		return fmt.Errorf("edge invalidation failed for thread %v: %w", t, err)
 	}
 	return txn.Commit()
 }
@@ -84,8 +91,17 @@ func (hb *dsHeadBook) SetHead(t thread.ID, p peer.ID, c cid.Cid) error {
 }
 
 func (hb *dsHeadBook) SetHeads(t thread.ID, p peer.ID, heads []cid.Cid) error {
-	key := dsLogKey(t, p, hbBase)
-	hr := pb.HeadBookRecord{}
+	txn, err := hb.ds.NewTransaction(false)
+	if err != nil {
+		return fmt.Errorf("error when creating txn in datastore: %w", err)
+	}
+	defer txn.Discard()
+
+	var (
+		hr  pb.HeadBookRecord
+		key = dsLogKey(t, p, hbBase)
+	)
+
 	for i := range heads {
 		if !heads[i].Defined() {
 			log.Warnf("ignoring head %s is undefined for %s", heads[i], key)
@@ -95,14 +111,14 @@ func (hb *dsHeadBook) SetHeads(t thread.ID, p peer.ID, heads []cid.Cid) error {
 		hr.Heads = append(hr.Heads, entry)
 	}
 
-	data, err := proto.Marshal(&hr)
-	if err != nil {
+	if data, err := proto.Marshal(&hr); err != nil {
 		return fmt.Errorf("error when marshaling headbookrecord proto for %v: %w", key, err)
-	}
-	if err = hb.ds.Put(key, data); err != nil {
+	} else if err = txn.Put(key, data); err != nil {
 		return fmt.Errorf("error when saving new head record in datastore for %v: %w", key, err)
+	} else if err := hb.invalidateEdge(txn, t); err != nil {
+		return fmt.Errorf("edge invalidation failed for thread %v: %w", t, err)
 	}
-	return nil
+	return txn.Commit()
 }
 
 func (hb *dsHeadBook) Heads(t thread.ID, p peer.ID) ([]cid.Cid, error) {
@@ -126,11 +142,73 @@ func (hb *dsHeadBook) Heads(t thread.ID, p peer.ID) ([]cid.Cid, error) {
 }
 
 func (hb *dsHeadBook) ClearHeads(t thread.ID, p peer.ID) error {
-	key := dsLogKey(t, p, hbBase)
-	if err := hb.ds.Delete(key); err != nil {
+	txn, err := hb.ds.NewTransaction(false)
+	if err != nil {
+		return fmt.Errorf("error when creating txn in datastore: %w", err)
+	}
+	defer txn.Discard()
+	var key = dsLogKey(t, p, hbBase)
+	if err := txn.Delete(key); err != nil {
 		return fmt.Errorf("error when deleting heads from %s", key)
+	} else if err := hb.invalidateEdge(txn, t); err != nil {
+		return fmt.Errorf("edge invalidation failed for thread %v: %w", t, err)
 	}
-	return nil
+	return txn.Commit()
+}
+
+func (hb *dsHeadBook) HeadsEdge(tid thread.ID) (uint64, error) {
+	txn, err := hb.ds.NewTransaction(false)
+	if err != nil {
+		return 0, fmt.Errorf("error when creating txn in datastore: %w", err)
+	}
+	defer txn.Discard()
+	edge, err := hb.getEdge(txn, tid)
+	if err != nil {
+		return 0, err
+	}
+	return edge, txn.Commit()
+}
+
+func (hb *dsHeadBook) getEdge(txn ds.Txn, tid thread.ID) (uint64, error) {
+	var key = dsThreadKey(tid, hbEdge)
+	if v, err := txn.Get(key); err == nil {
+		return binary.BigEndian.Uint64(v), nil
+	} else if err != ds.ErrNotFound {
+		return 0, err
+	}
+
+	// edge not evaluated/invalidated, let's compute it
+	result, err := txn.Query(query.Query{Prefix: dsThreadKey(tid, hbBase).String(), KeysOnly: false})
+	if err != nil {
+		return 0, err
+	}
+	defer result.Close()
+
+	var hs []util.LogHead
+	for entry := range result.Next() {
+		_, lid, heads, err := hb.decodeHeadEntry(entry, true)
+		if err != nil {
+			return 0, err
+		}
+		for i := 0; i < len(heads); i++ {
+			hs = append(hs, util.LogHead{Head: heads[i], LogID: lid})
+		}
+	}
+	if len(hs) == 0 {
+		return 0, core.ErrThreadNotFound
+	}
+
+	var (
+		buff [8]byte
+		edge = util.ComputeHeadsEdge(hs)
+	)
+	binary.BigEndian.PutUint64(buff[:], edge)
+	return edge, txn.Put(key, buff[:])
+}
+
+func (hb *dsHeadBook) invalidateEdge(txn ds.Txn, tid thread.ID) error {
+	var key = dsThreadKey(tid, hbEdge)
+	return txn.Delete(key)
 }
 
 // Dump entire headbook into the tree-structure.
@@ -182,37 +260,9 @@ func (hb *dsHeadBook) traverse(withHeads bool) (map[thread.ID]map[peer.ID][]cid.
 	defer result.Close()
 
 	for entry := range result.Next() {
-		kns := ds.RawKey(entry.Key).Namespaces()
-		if len(kns) < 3 {
-			return nil, fmt.Errorf("bad headbook key detected: %s", entry.Key)
-		}
-
-		// get thread and log IDs from the key components
-		ts, ls := kns[len(kns)-2], kns[len(kns)-1]
-
-		// parse thread ID
-		tid, err := parseThreadID(ts)
+		tid, lid, heads, err := hb.decodeHeadEntry(entry, withHeads)
 		if err != nil {
-			return nil, fmt.Errorf("cannot restore thread ID %s: %w", ts, err)
-		}
-
-		// parse log ID
-		lid, err := parseLogID(ls)
-		if err != nil {
-			return nil, fmt.Errorf("cannot restore log ID %s: %w", ls, err)
-		}
-
-		var heads []cid.Cid
-		if withHeads {
-			var hr pb.HeadBookRecord
-			if err := proto.Unmarshal(entry.Value, &hr); err != nil {
-				return nil, fmt.Errorf("cannot decode headbook record: %w", err)
-			}
-
-			heads = make([]cid.Cid, len(hr.Heads))
-			for i := range hr.Heads {
-				heads[i] = hr.Heads[i].Cid.Cid
-			}
+			return nil, err
 		}
 
 		lh, exist := data[tid]
@@ -226,3 +276,36 @@ func (hb *dsHeadBook) traverse(withHeads bool) (map[thread.ID]map[peer.ID][]cid.
 
 	return data, nil
 }
+
+func (hb *dsHeadBook) decodeHeadEntry(
+	entry query.Result,
+	withHeads bool,
+) (tid thread.ID, lid peer.ID, heads []cid.Cid, err error) {
+	kns := ds.RawKey(entry.Key).Namespaces()
+	if len(kns) < 4 {
+		err = fmt.Errorf("bad headbook key detected: %s", entry.Key)
+		return
+	}
+	// get thread and log IDs from the key components
+	var ts, ls = kns[len(kns)-2], kns[len(kns)-1]
+	if tid, err = parseThreadID(ts); err != nil {
+		err = fmt.Errorf("cannot restore thread ID %s: %w", ts, err)
+		return
+	}
+	if lid, err = parseLogID(ls); err != nil {
+		err = fmt.Errorf("cannot restore log ID %s: %w", ls, err)
+		return
+	}
+	if withHeads {
+		var hr pb.HeadBookRecord
+		if err = proto.Unmarshal(entry.Value, &hr); err != nil {
+			err = fmt.Errorf("cannot decode headbook record: %w", err)
+			return
+		}
+		heads = make([]cid.Cid, len(hr.Heads))
+		for i := range hr.Heads {
+			heads[i] = hr.Heads[i].Cid.Cid
+		}
+	}
+	return
+}
diff --git a/logstore/lstorehybrid/logstore.go b/logstore/lstorehybrid/logstore.go
index 2a26491..1f4e9d8 100644
--- a/logstore/lstorehybrid/logstore.go
+++ b/logstore/lstorehybrid/logstore.go
@@ -278,6 +278,10 @@ func (l *lstore) ClearHeads(tid thread.ID, lid peer.ID) error {
 	return l.inMem.ClearHeads(tid, lid)
 }
 
+func (l *lstore) HeadsEdge(tid thread.ID) (uint64, error) {
+	return l.inMem.HeadsEdge(tid)
+}
+
 func (l *lstore) Threads() (thread.IDSlice, error) {
 	return l.inMem.Threads()
 }
diff --git a/logstore/lstoremem/headbook.go b/logstore/lstoremem/headbook.go
index 8d77857..315824c 100644
--- a/logstore/lstoremem/headbook.go
+++ b/logstore/lstoremem/headbook.go
@@ -7,28 +7,42 @@ import (
 	"github.com/libp2p/go-libp2p-core/peer"
 	core "github.com/textileio/go-threads/core/logstore"
 	"github.com/textileio/go-threads/core/thread"
+	"github.com/textileio/go-threads/util"
 )
 
 type memoryHeadBook struct {
 	sync.RWMutex
-
-	heads map[thread.ID]map[peer.ID]map[cid.Cid]struct{}
+	threads map[thread.ID]struct {
+		heads map[peer.ID]map[cid.Cid]struct{}
+		edge  uint64
+	}
 }
 
-func (mhb *memoryHeadBook) getHeads(t thread.ID, p peer.ID) (map[cid.Cid]struct{}, bool) {
-	lmap, found := mhb.heads[t]
-	if lmap == nil {
-		return nil, found
+func (mhb *memoryHeadBook) getHeads(t thread.ID, p peer.ID, createEmpty bool) map[cid.Cid]struct{} {
+	lmap := mhb.threads[t]
+	if lmap.heads == nil {
+		if !createEmpty {
+			return nil
+		}
+		lmap.heads = make(map[peer.ID]map[cid.Cid]struct{}, 1)
+		mhb.threads[t] = lmap
+	}
+	hmap := lmap.heads[p]
+	if hmap == nil && createEmpty {
+		hmap = make(map[cid.Cid]struct{})
+		lmap.heads[p] = hmap
 	}
-	hmap, found := lmap[p]
-	return hmap, found
+	return hmap
 }
 
 var _ core.HeadBook = (*memoryHeadBook)(nil)
 
 func NewHeadBook() core.HeadBook {
 	return &memoryHeadBook{
-		heads: map[thread.ID]map[peer.ID]map[cid.Cid]struct{}{},
+		threads: make(map[thread.ID]struct {
+			heads map[peer.ID]map[cid.Cid]struct{}
+			edge  uint64
+		}),
 	}
 }
 
@@ -39,16 +53,9 @@ func (mhb *memoryHeadBook) AddHead(t thread.ID, p peer.ID, head cid.Cid) error {
 func (mhb *memoryHeadBook) AddHeads(t thread.ID, p peer.ID, heads []cid.Cid) error {
 	mhb.Lock()
 	defer mhb.Unlock()
+	defer mhb.updateEdge(t)
 
-	hmap, _ := mhb.getHeads(t, p)
-	if hmap == nil {
-		if mhb.heads[t] == nil {
-			mhb.heads[t] = make(map[peer.ID]map[cid.Cid]struct{}, 1)
-		}
-		hmap = make(map[cid.Cid]struct{}, len(heads))
-		mhb.heads[t][p] = hmap
-	}
-
+	hmap := mhb.getHeads(t, p, true)
 	for _, h := range heads {
 		if !h.Defined() {
 			log.Warnf("was passed nil head for %s", p)
@@ -66,23 +73,24 @@ func (mhb *memoryHeadBook) SetHead(t thread.ID, p peer.ID, head cid.Cid) error {
 func (mhb *memoryHeadBook) SetHeads(t thread.ID, p peer.ID, heads []cid.Cid) error {
 	mhb.Lock()
 	defer mhb.Unlock()
+	defer mhb.updateEdge(t)
 
-	hmap, _ := mhb.getHeads(t, p)
-	if hmap == nil {
-		if mhb.heads[t] == nil {
-			mhb.heads[t] = make(map[peer.ID]map[cid.Cid]struct{}, 1)
-		}
-	}
-	hmap = make(map[cid.Cid]struct{}, len(heads))
-	mhb.heads[t][p] = hmap
-
+	var hset = make(map[cid.Cid]struct{}, len(heads))
 	for _, h := range heads {
 		if !h.Defined() {
 			log.Warnf("was passed nil head for %s", p)
 			continue
 		}
-		hmap[h] = struct{}{}
+		hset[h] = struct{}{}
 	}
+	// replace heads
+	if mhb.threads[t].heads == nil {
+		mhb.threads[t] = struct {
+			heads map[peer.ID]map[cid.Cid]struct{}
+			edge  uint64
+		}{heads: make(map[peer.ID]map[cid.Cid]struct{})}
+	}
+	mhb.threads[t].heads[p] = hset
 	return nil
 }
 
@@ -90,12 +98,13 @@ func (mhb *memoryHeadBook) Heads(t thread.ID, p peer.ID) ([]cid.Cid, error) {
 	mhb.RLock()
 	defer mhb.RUnlock()
 
-	var heads []cid.Cid
-	hmap, _ := mhb.getHeads(t, p)
-	if hmap == nil {
-		return heads, nil
+	hset := mhb.getHeads(t, p, false)
+	if hset == nil {
+		return nil, nil
 	}
-	for h := range hmap {
+
+	var heads = make([]cid.Cid, 0, len(hset))
+	for h := range hset {
 		heads = append(heads, h)
 	}
 	return heads, nil
@@ -105,24 +114,57 @@ func (mhb *memoryHeadBook) ClearHeads(t thread.ID, p peer.ID) error {
 	mhb.Lock()
 	defer mhb.Unlock()
 
-	lmap := mhb.heads[t]
-	if lmap != nil {
-		delete(lmap, p)
-		if len(lmap) == 0 {
-			delete(mhb.heads, t)
-		}
+	var lset = mhb.threads[t].heads
+	if lset == nil {
+		return nil
+	}
+	delete(lset, p)
+	if len(lset) == 0 {
+		delete(mhb.threads, t)
+	} else {
+		mhb.updateEdge(t)
 	}
 	return nil
 }
 
+func (mhb *memoryHeadBook) HeadsEdge(t thread.ID) (uint64, error) {
+	mhb.RLock()
+	defer mhb.RUnlock()
+
+	lset, found := mhb.threads[t]
+	if !found {
+		return 0, core.ErrThreadNotFound
+	}
+	// invariant: edge always precomputed for existing thread
+	return lset.edge, nil
+}
+
+func (mhb *memoryHeadBook) updateEdge(t thread.ID) {
+	// invariant: requested thread exist
+	var (
+		lset  = mhb.threads[t]
+		heads = make([]util.LogHead, 0, len(lset.heads))
+	)
+	for lid, hs := range lset.heads {
+		for head := range hs {
+			heads = append(heads, util.LogHead{
+				LogID: lid,
+				Head:  head,
+			})
+		}
+	}
+	lset.edge = util.ComputeHeadsEdge(heads)
+	mhb.threads[t] = lset
+}
+
 func (mhb *memoryHeadBook) DumpHeads() (core.DumpHeadBook, error) {
 	var dump = core.DumpHeadBook{
-		Data: make(map[thread.ID]map[peer.ID][]cid.Cid, len(mhb.heads)),
+		Data: make(map[thread.ID]map[peer.ID][]cid.Cid, len(mhb.threads)),
 	}
 
-	for tid, logs := range mhb.heads {
-		lm := make(map[peer.ID][]cid.Cid, len(logs))
-		for lid, hs := range logs {
+	for tid, lset := range mhb.threads {
+		lm := make(map[peer.ID][]cid.Cid, len(lset.heads))
+		for lid, hs := range lset.heads {
 			heads := make([]cid.Cid, 0, len(hs))
 			for head := range hs {
 				heads = append(heads, head)
@@ -140,7 +182,12 @@ func (mhb *memoryHeadBook) RestoreHeads(dump core.DumpHeadBook) error {
 		return core.ErrEmptyDump
 	}
 
-	var restored = make(map[thread.ID]map[peer.ID]map[cid.Cid]struct{}, len(dump.Data))
+	// reset stored
+	mhb.threads = make(map[thread.ID]struct {
+		heads map[peer.ID]map[cid.Cid]struct{}
+		edge  uint64
+	}, len(dump.Data))
+
 	for tid, logs := range dump.Data {
 		lm := make(map[peer.ID]map[cid.Cid]struct{}, len(logs))
 		for lid, hs := range logs {
@@ -150,9 +197,12 @@ func (mhb *memoryHeadBook) RestoreHeads(dump core.DumpHeadBook) error {
 			}
 			lm[lid] = hm
 		}
-		restored[tid] = lm
+		mhb.threads[tid] = struct {
+			heads map[peer.ID]map[cid.Cid]struct{}
+			edge  uint64
+		}{heads: lm}
+		mhb.updateEdge(tid)
 	}
 
-	mhb.heads = restored
 	return nil
 }
diff --git a/net/server.go b/net/server.go
index b9e3e11..c17633c 100644
--- a/net/server.go
+++ b/net/server.go
@@ -16,6 +16,7 @@ import (
 	lstore "github.com/textileio/go-threads/core/logstore"
 	"github.com/textileio/go-threads/core/thread"
 	pb "github.com/textileio/go-threads/net/pb"
+	"github.com/textileio/go-threads/util"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 )
@@ -157,11 +158,18 @@ func (s *server) GetRecords(ctx context.Context, req *pb.GetRecordsRequest) (*pb
 	}
 	log.Debugf("received get records request from %s", pid)
 
-	pbrecs := &pb.GetRecordsReply{}
+	var pbrecs = &pb.GetRecordsReply{}
 	if err := s.checkServiceKey(req.Body.ThreadID.ID, req.Body.ServiceKey); err != nil {
 		return pbrecs, err
 	}
 
+	// fast check if requested offsets are equal with thread heads
+	if changed, err := s.headsChanged(req); err != nil {
+		return nil, err
+	} else if !changed {
+		return pbrecs, nil
+	}
+
 	reqd := make(map[peer.ID]*pb.GetRecordsRequest_Body_LogEntry)
 	for _, l := range req.Body.Logs {
 		reqd[l.LogID.ID] = l
@@ -172,16 +180,24 @@ func (s *server) GetRecords(ctx context.Context, req *pb.GetRecordsRequest) (*pb
 	}
 	pbrecs.Logs = make([]*pb.GetRecordsReply_LogEntry, len(info.Logs))
 
+	var recordLimit = MaxPullLimit
 	for i, lg := range info.Logs {
-		var offset cid.Cid
-		var limit int
-		var pblg *pb.Log
+		var (
+			offset cid.Cid
+			limit  int
+			pblg   *pb.Log
+
+			// correct records-per-log limit
+			logsRemain     = len(info.Logs) - i
+			logRecordLimit = recordLimit / logsRemain
+		)
+
 		if opts, ok := reqd[lg.ID]; ok {
 			offset = opts.Offset.Cid
-			limit = int(opts.Limit)
+			limit = minInt(int(opts.Limit), logRecordLimit)
 		} else {
 			offset = cid.Undef
-			limit = MaxPullLimit
+			limit = logRecordLimit
 			pblg = logToProto(lg)
 		}
 		recs, err := s.net.getLocalRecords(ctx, req.Body.ThreadID.ID, lg.ID, offset, limit)
@@ -202,6 +218,9 @@ func (s *server) GetRecords(ctx context.Context, req *pb.GetRecordsRequest) (*pb
 		}
 		pbrecs.Logs[i] = entry
 
+		// update remaining records limit
+		recordLimit -= len(recs)
+
 		log.Debugf("sending %d records in log %s to %s", len(recs), lg.ID, pid)
 	}
 
@@ -266,6 +285,19 @@ func (s *server) checkServiceKey(id thread.ID, k *pb.ProtoKey) error {
 	return nil
 }
 
+// headsChanged determines if thread heads are different from the requested offsets.
+func (s *server) headsChanged(req *pb.GetRecordsRequest) (bool, error) {
+	var reqHeads = make([]util.LogHead, len(req.Body.Logs))
+	for i, l := range req.Body.GetLogs() {
+		reqHeads[i] = util.LogHead{Head: l.Offset.Cid, LogID: l.LogID.ID}
+	}
+	var currEdge, err = s.net.store.HeadsEdge(req.Body.ThreadID.ID)
+	if err != nil {
+		return false, err
+	}
+	return util.ComputeHeadsEdge(reqHeads) != currEdge, nil
+}
+
 // verifyRequest verifies that the signature associated with a request is valid.
 func verifyRequest(header *pb.Header, body proto.Marshaler) (pid peer.ID, err error) {
 	if header == nil || body == nil {
@@ -325,3 +357,10 @@ func addrsFromProto(pa []pb.ProtoAddr) []ma.Multiaddr {
 	}
 	return mas
 }
+
+func minInt(x, y int) int {
+	if x < y {
+		return x
+	}
+	return y
+}
diff --git a/util/util.go b/util/util.go
index b90557e..cf625ce 100644
--- a/util/util.go
+++ b/util/util.go
@@ -4,12 +4,15 @@ import (
 	"crypto/rand"
 	"encoding/json"
 	"fmt"
+	"hash/fnv"
 	"os"
 	"path/filepath"
+	"sort"
 
 	"github.com/alecthomas/jsonschema"
 	"github.com/dgraph-io/badger/options"
 	ipfslite "github.com/hsanjuan/ipfs-lite"
+	"github.com/ipfs/go-cid"
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/libp2p/go-libp2p-core/peer"
 	ma "github.com/multiformats/go-multiaddr"
@@ -196,3 +199,23 @@ func MakeToken(n int) string {
 	}
 	return encoded
 }
+
+type LogHead struct {
+	LogID peer.ID
+	Head  cid.Cid
+}
+
+func ComputeHeadsEdge(hs []LogHead) uint64 {
+	// sort heads for deterministic edge computation
+	sort.Slice(hs, func(i, j int) bool {
+		if hs[i].LogID == hs[j].LogID {
+			return hs[i].Head.KeyString() < hs[j].Head.KeyString()
+		}
+		return hs[i].LogID < hs[j].LogID
+	})
+	hasher := fnv.New64a()
+	for i := 0; i < len(hs); i++ {
+		_, _ = hasher.Write(hs[i].Head.Bytes())
+	}
+	return hasher.Sum64()
+}


Test patch:
diff --git a/db/collection_test.go b/db/collection_test.go
index 2a36694..6dd37ab 100644
--- a/db/collection_test.go
+++ b/db/collection_test.go
@@ -872,6 +872,9 @@ func TestModifiedSince(t *testing.T) {
 		})
 		checkErr(t, err)
 
+		// let write transaction propagate
+		time.Sleep(10 * time.Millisecond)
+
 		var instance []byte
 		err = c.ReadTxn(func(txn *Txn) (err error) {
 			instance, err = txn.FindByID(res[0])
@@ -891,7 +894,7 @@ func TestModifiedSince(t *testing.T) {
 			t.Fatalf("should have had %d modified instance", 1)
 		}
 		if mods[0] != p.ID {
-			t.Fatalf("should have modfied id %s", p.ID)
+			t.Fatalf("should have modified id %s", p.ID)
 		}
 	})
 	t.Run("WithCreateSaveAndDelete", func(t *testing.T) {
@@ -937,6 +940,9 @@ func TestModifiedSince(t *testing.T) {
 		})
 		checkErr(t, err)
 
+		// let write transactions propagate
+		time.Sleep(10 * time.Millisecond)
+
 		var mods []core.InstanceID
 		err = c.ReadTxn(func(txn *Txn) (err error) {
 			mods, err = txn.ModifiedSince(before)
diff --git a/db/query_more_test.go b/db/query_more_test.go
index c72f18a..5c54e05 100644
--- a/db/query_more_test.go
+++ b/db/query_more_test.go
@@ -89,6 +89,11 @@ var (
 
 		{name: "SortAllAscFloat", query: OrderBy("Meta.Rating"), resIdx: []int{0, 1, 2, 3, 4}, ordered: true},
 		{name: "SortAllDescFloat", query: OrderByDesc("Meta.Rating"), resIdx: []int{4, 3, 2, 1, 0}, ordered: true},
+
+		{name: "LimitTotalReadsOutside", query: Where("Meta.TotalReads").Gt(float64(100)).LimitTo(2), resIdx: []int{3, 4}},
+		{name: "LimitTotalReadsInside", query: Where("Meta.TotalReads").Lt(float64(100)).LimitTo(2), resIdx: []int{0, 1}},
+
+		{name: "LimitWithSkip", query: Where("Meta.TotalReads").Gt(float64(0)).SkipNum(1).LimitTo(3), resIdx: []int{1, 2, 3}},
 	}
 )
 
diff --git a/test/addr_book_suite.go b/test/addr_book_suite.go
index 48a93b4..344d7a4 100644
--- a/test/addr_book_suite.go
+++ b/test/addr_book_suite.go
@@ -125,8 +125,8 @@ func testAddAddress(ab core.AddrBook) func(*testing.T) {
 			// 1 second left
 			check(t, ab.AddAddrs(tid, id, addrs, 3*time.Second))
 			// 3 seconds left
-			time.Sleep(2 * time.Second)
-			// 1 seconds left.
+			time.Sleep(1 * time.Second)
+			// 2 seconds left.
 
 			// We still have the address.
 			AssertAddressesEqual(t, addrs, checkedAddrs(t, ab, tid, id))
diff --git a/test/headbook_suite.go b/test/headbook_suite.go
index c485d4a..cc638da 100644
--- a/test/headbook_suite.go
+++ b/test/headbook_suite.go
@@ -12,6 +12,7 @@ import (
 	mh "github.com/multiformats/go-multihash"
 	core "github.com/textileio/go-threads/core/logstore"
 	"github.com/textileio/go-threads/core/thread"
+	"github.com/textileio/go-threads/util"
 )
 
 var headBookSuite = map[string]func(hb core.HeadBook) func(*testing.T){
@@ -19,6 +20,7 @@ var headBookSuite = map[string]func(hb core.HeadBook) func(*testing.T){
 	"SetGetHeads": testHeadBookSetHeads,
 	"ClearHeads":  testHeadBookClearHeads,
 	"ExportHeads": testHeadBookExport,
+	"HeadsEdge":   testHeadBookEdge,
 }
 
 type HeadBookFactory func() (core.HeadBook, func())
@@ -148,6 +150,90 @@ func testHeadBookClearHeads(hb core.HeadBook) func(t *testing.T) {
 	}
 }
 
+func testHeadBookEdge(hb core.HeadBook) func(t *testing.T) {
+	return func(t *testing.T) {
+		var (
+			numLogs  = 3
+			numHeads = 2
+
+			genHeadSets = func(cases int) (thread.ID, []map[peer.ID][]cid.Cid) {
+				tid, logs := genHeads(numLogs, cases*numHeads)
+				splitted := make([]map[peer.ID][]cid.Cid, cases)
+				for i := 0; i < cases; i++ {
+					hs := make(map[peer.ID][]cid.Cid)
+					for lid, cids := range logs {
+						hs[lid] = append(hs[lid], cids[i*numHeads:(i+1)*numHeads]...)
+					}
+					splitted[i] = hs
+				}
+				return tid, splitted
+			}
+
+			equalEdges = func(edge uint64, logs map[peer.ID][]cid.Cid) bool {
+				var heads []util.LogHead
+				for lid, cids := range logs {
+					for _, c := range cids {
+						heads = append(heads, util.LogHead{LogID: lid, Head: c})
+					}
+				}
+				return util.ComputeHeadsEdge(heads) == edge
+			}
+
+			// generate 3 sets of heads
+			tid, heads          = genHeadSets(3)
+			hSet1, hSet2, hSet3 = heads[0], heads[1], heads[2]
+		)
+
+		if _, err := hb.HeadsEdge(tid); err != core.ErrThreadNotFound {
+			t.Error("expected to get error on retrieving non-existing thread's edge")
+		}
+		for lid, hs := range hSet1 {
+			if stored, err := hb.Heads(tid, lid); err != nil || len(stored) > 0 {
+				t.Error("expected heads to be empty on init without errors")
+			}
+			if err := hb.SetHeads(tid, lid, hs); err != nil {
+				t.Fatalf("error when adding heads: %v", err)
+			}
+		}
+
+		edge1, err := hb.HeadsEdge(tid)
+		if err != nil {
+			t.Errorf("error while getting thread's edge: %v", err)
+		}
+
+		// set new heads
+		for lid, hs := range hSet2 {
+			if err := hb.SetHeads(tid, lid, hs); err != nil {
+				t.Fatalf("error when adding heads: %v", err)
+			}
+		}
+		edge2, err := hb.HeadsEdge(tid)
+		if err != nil {
+			t.Errorf("error while getting thread's edge: %v", err)
+		}
+		if edge1 == edge2 {
+			t.Error("edges should not be equal after setting new heads")
+		}
+		if !equalEdges(edge2, hSet2) {
+			t.Error("stored and manually computed edges doesn't match")
+		}
+
+		// add some more heads
+		for lid, hs := range hSet3 {
+			if err := hb.AddHeads(tid, lid, hs); err != nil {
+				t.Fatalf("error when adding heads: %v", err)
+			}
+		}
+		edge3, err := hb.HeadsEdge(tid)
+		if err != nil {
+			t.Errorf("error while getting thread's edge: %v", err)
+		}
+		if edge3 == edge1 || edge3 == edge2 {
+			t.Error("edges should not be equal after setting/adding new heads")
+		}
+	}
+}
+
 func testHeadBookExport(hb core.HeadBook) func(t *testing.T) {
 	return func(t *testing.T) {
 		var (