Repository: textileio/go-threads
Pull request #473 (2020-12-11 16:47:06)

mod: update ds-mongo
Chore... update to release vs. hash.


Patch:
diff --git a/db/collection.go b/db/collection.go
index 096f2ff..3b5b67a 100644
--- a/db/collection.go
+++ b/db/collection.go
@@ -16,7 +16,6 @@ import (
 	ds "github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/query"
 	format "github.com/ipfs/go-ipld-format"
-	dse "github.com/textileio/go-datastore-extensions"
 	"github.com/textileio/go-threads/core/app"
 	core "github.com/textileio/go-threads/core/db"
 	"github.com/textileio/go-threads/core/thread"
@@ -282,51 +281,12 @@ func (f filter) Filter(e query.Entry) bool {
 }
 
 // ModifiedSince returns a list of all instances that have been modified (and/or touched) since `time`.
-func (c *Collection) ModifiedSince(time int64, opts ...TxnOption) (modified []core.InstanceID, err error) {
-	// Need to identify instances that have been touched (created, saved, or deleted) since `time`.
-	// The _mod field tracks modified instances, but not those that have been deleted, so we need
-	// to query the dispatcher for all (unique) instances in this collection that have been modified
-	// at all since `time`.
-	c.db.dispatcher.Lock().Lock()
-	defer c.db.dispatcher.Lock().Unlock()
-
-	txn, err := c.db.dispatcher.Store().NewTransactionExtended(false)
-	if err != nil {
-		return nil, err
-	}
-	defer txn.Discard()
-
-	results, err := txn.QueryExtended(dse.QueryExt{
-		Query: query.Query{
-			Prefix: dsDispatcherPrefix.String(),
-			Filters: []query.Filter{
-				filter{
-					Collection: c.name,
-				},
-			},
-			KeysOnly: true,
-		},
-		SeekPrefix: dsDispatcherPrefix.ChildString(strconv.FormatInt(time, 10)).String(),
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	type void struct{}
-	var member void
-	set := make(map[core.InstanceID]void)
-
-	for r := range results.Next() {
-		if r.Error != nil {
-			return nil, r.Error
-		}
-		id := ds.NewKey(r.Key)
-		set[core.InstanceID(id.Name())] = member
-	}
-	for k := range set {
-		modified = append(modified, k)
-	}
-	return modified, nil
+func (c *Collection) ModifiedSince(time int64, opts ...TxnOption) (ids []core.InstanceID, err error) {
+	_ = c.ReadTxn(func(txn *Txn) error {
+		ids, err = txn.ModifiedSince(time)
+		return err
+	}, opts...)
+	return
 }
 
 // validInstance validates the json object against the collection schema.
diff --git a/db/dispatcher.go b/db/dispatcher.go
index 6a6f96e..6e21b9d 100644
--- a/db/dispatcher.go
+++ b/db/dispatcher.go
@@ -48,10 +48,6 @@ func (d *dispatcher) Store() kt.TxnDatastoreExtended {
 	return d.store
 }
 
-func (d *dispatcher) Lock() *sync.RWMutex {
-	return &d.lock
-}
-
 // Register takes a reducer to be invoked with each dispatched event.
 func (d *dispatcher) Register(reducer Reducer) {
 	d.lock.Lock()
diff --git a/db/manager.go b/db/manager.go
index 936ba40..4c1b48d 100644
--- a/db/manager.go
+++ b/db/manager.go
@@ -69,6 +69,9 @@ func NewManager(store kt.TxnDatastoreExtended, network app.Net, opts ...NewOptio
 	defer results.Close()
 	invalids := make(map[thread.ID]struct{})
 	for res := range results.Next() {
+		if res.Error != nil {
+			return nil, err
+		}
 		parts := strings.Split(ds.RawKey(res.Key).String(), "/")
 		if len(parts) < 3 {
 			continue
diff --git a/db/query.go b/db/query.go
index 6b06041..28f030e 100644
--- a/db/query.go
+++ b/db/query.go
@@ -5,8 +5,12 @@ import (
 	"fmt"
 	"reflect"
 	"sort"
+	"strconv"
 	"strings"
 
+	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/query"
+	dse "github.com/textileio/go-datastore-extensions"
 	core "github.com/textileio/go-threads/core/db"
 )
 
@@ -218,32 +222,32 @@ func (q *Query) SkipNum(num int) *Query {
 
 // Criterion helpers
 
-// Eq is an equality operator against a field
+// Eq is an equality operator against a field.
 func (c *Criterion) Eq(value interface{}) *Query {
 	return c.createcriterion(Eq, value)
 }
 
-// Ne is a not equal operator against a field
+// Ne is a not equal operator against a field.
 func (c *Criterion) Ne(value interface{}) *Query {
 	return c.createcriterion(Ne, value)
 }
 
-// Gt is a greater operator against a field
+// Gt is a greater operator against a field.
 func (c *Criterion) Gt(value interface{}) *Query {
 	return c.createcriterion(Gt, value)
 }
 
-// Lt is a less operation against a field
+// Lt is a less operation against a field.
 func (c *Criterion) Lt(value interface{}) *Query {
 	return c.createcriterion(Lt, value)
 }
 
-// Ge is a greater or equal operator against a field
+// Ge is a greater or equal operator against a field.
 func (c *Criterion) Ge(value interface{}) *Query {
 	return c.createcriterion(Ge, value)
 }
 
-// Le is a less or equal operator against a field
+// Le is a less or equal operator against a field.
 func (c *Criterion) Le(value interface{}) *Query {
 	return c.createcriterion(Le, value)
 }
@@ -286,7 +290,7 @@ func (c *Criterion) createcriterion(op Operation, value interface{}) *Query {
 	return c.query
 }
 
-// Find queries for instances by Query
+// Find queries for instances by Query.
 func (t *Txn) Find(q *Query) ([][]byte, error) {
 	if err := t.collection.db.connector.Validate(t.token, true); err != nil {
 		return nil, err
@@ -478,3 +482,49 @@ func traverseFieldPathMap(value map[string]interface{}, fieldPath string) (refle
 	}
 	return reflect.ValueOf(curr), nil
 }
+
+// ModifiedSince returns a list of all instances that have been modified (and/or touched) since `time`.
+// The _mod field tracks modified instances, but not those that have been deleted, so we need
+// to query the dispatcher for all (unique) instances in this collection that have been modified
+// at all since `time`.
+func (t *Txn) ModifiedSince(time int64) (ids []core.InstanceID, err error) {
+	txn, err := t.collection.db.datastore.NewTransactionExtended(true)
+	if err != nil {
+		return nil, err
+	}
+	defer txn.Discard()
+
+	timestr := strconv.FormatInt(time, 10)
+	res, err := txn.QueryExtended(dse.QueryExt{
+		Query: query.Query{
+			Prefix: dsDispatcherPrefix.String(),
+			Filters: []query.Filter{
+				filter{
+					Collection: t.collection.name,
+				},
+			},
+			KeysOnly: true,
+		},
+		SeekPrefix: dsDispatcherPrefix.ChildString(timestr).String(),
+	})
+	if err != nil {
+		return nil, err
+	}
+	defer res.Close()
+
+	set := make(map[core.InstanceID]struct{})
+	for r := range res.Next() {
+		if r.Error != nil {
+			return nil, r.Error
+		}
+		id := ds.NewKey(r.Key)
+		set[core.InstanceID(id.Name())] = struct{}{}
+	}
+	ids = make([]core.InstanceID, len(set))
+	i := 0
+	for k := range set {
+		ids[i] = k
+		i++
+	}
+	return ids, nil
+}
diff --git a/docker-compose-dev.yml b/docker-compose-dev.yml
index a6915e5..f7ac4a5 100644
--- a/docker-compose-dev.yml
+++ b/docker-compose-dev.yml
@@ -24,8 +24,6 @@ services:
       - mongo
   mongo:
     image: mongo:latest
-    expose:
-      - "27017"
     ports:
       - "127.0.0.1:27017:27017"
     command:
diff --git a/docker-compose.yml b/docker-compose.yml
index ffdbeee..9118ae0 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -29,8 +29,6 @@ services:
     restart: always
     volumes:
       - "${THRDS_REPO}/mongodb:/data/mongodb"
-    expose:
-      - "27017"
     ports:
       - "127.0.0.1:27017:27017"
     command:
diff --git a/go.mod b/go.mod
index 6c72831..84ed0d5 100644
--- a/go.mod
+++ b/go.mod
@@ -60,7 +60,7 @@ require (
 	github.com/smartystreets/goconvey v0.0.0-20190710185942-9d28bd7c0945 // indirect
 	github.com/textileio/go-datastore-extensions v1.0.1
 	github.com/textileio/go-ds-badger v0.2.7-0.20201204225019-4ee78c4a40e2
-	github.com/textileio/go-ds-mongo v0.1.4-0.20201205032204-63f494d9c3f3
+	github.com/textileio/go-ds-mongo v0.1.4
 	github.com/tidwall/gjson v1.3.5
 	github.com/tidwall/sjson v1.0.4
 	github.com/whyrusleeping/base32 v0.0.0-20170828182744-c30ac30633cc
diff --git a/go.sum b/go.sum
index 826ed5c..f534a0e 100644
--- a/go.sum
+++ b/go.sum
@@ -945,8 +945,8 @@ github.com/textileio/go-datastore-extensions v1.0.1 h1:qIJGqJaigQ1wD4TdwS/hf73u0
 github.com/textileio/go-datastore-extensions v1.0.1/go.mod h1:Pzj9FDRkb55910dr/FX8M7WywvnS26gBgEDez1ZBuLE=
 github.com/textileio/go-ds-badger v0.2.7-0.20201204225019-4ee78c4a40e2 h1:VIWkccJp5k8x1hbwQsSn6Xnns4wUDEW6XdiMLiOpr7k=
 github.com/textileio/go-ds-badger v0.2.7-0.20201204225019-4ee78c4a40e2/go.mod h1:qEZ/z1KyoRhGS5MYEbIcWUCCPd/0HxCkFDVeJgP1RcI=
-github.com/textileio/go-ds-mongo v0.1.4-0.20201205032204-63f494d9c3f3 h1:agZyjIQP6gqAcKxp06bAM39iAYd84kfkGJ8XyY3hNQI=
-github.com/textileio/go-ds-mongo v0.1.4-0.20201205032204-63f494d9c3f3/go.mod h1:Zf6JlMPiIQUUmGlFFn5Z65C9p9LAvPg7XvX+qdGmTsU=
+github.com/textileio/go-ds-mongo v0.1.4 h1:5PPMEmqCfeCySzadtuBXMEJVDGP1Cga/CQXo3GWetmE=
+github.com/textileio/go-ds-mongo v0.1.4/go.mod h1:Zf6JlMPiIQUUmGlFFn5Z65C9p9LAvPg7XvX+qdGmTsU=
 github.com/tidwall/gjson v1.3.5 h1:2oW9FBNu8qt9jy5URgrzsVx/T/KSn3qn/smJQ0crlDQ=
 github.com/tidwall/gjson v1.3.5/go.mod h1:P256ACg0Mn+j1RXIDXoss50DeIABTYK1PULOJHhxOls=
 github.com/tidwall/match v1.0.1 h1:PnKP62LPNxHKTwvHHZZzdOAOCtsJTjo6dZLCwpKm5xc=


Test patch:
diff --git a/broadcast/broadcast_test.go b/broadcast/broadcast_test.go
index 435bd44..fd79fea 100644
--- a/broadcast/broadcast_test.go
+++ b/broadcast/broadcast_test.go
@@ -11,7 +11,7 @@ import (
 const (
 	N       = 3
 	testStr = "Test"
-	timeout = time.Second
+	timeout = time.Second * 2
 )
 
 type ListenFunc func(int, *Broadcaster, *sync.WaitGroup)
diff --git a/db/collection_test.go b/db/collection_test.go
index 51fd005..2a36694 100644
--- a/db/collection_test.go
+++ b/db/collection_test.go
@@ -862,7 +862,7 @@ func TestModifiedSince(t *testing.T) {
 		})
 		checkErr(t, err)
 
-		before := time.Now().UnixNano()
+		before := time.Now().Add(-time.Second).UnixNano()
 
 		newPerson := util.JSONFromInstance(Person{Name: "Alice", Age: 42})
 		var res []core.InstanceID
@@ -872,13 +872,21 @@ func TestModifiedSince(t *testing.T) {
 		})
 		checkErr(t, err)
 
-		instance, err := c.FindByID(res[0])
+		var instance []byte
+		err = c.ReadTxn(func(txn *Txn) (err error) {
+			instance, err = txn.FindByID(res[0])
+			return
+		})
 		checkErr(t, err)
-
 		p := &Person{}
 		util.InstanceFromJSON(instance, p)
 
-		mods, err := c.ModifiedSince(before)
+		var mods []core.InstanceID
+		err = c.ReadTxn(func(txn *Txn) (err error) {
+			mods, err = txn.ModifiedSince(before)
+			return
+		})
+		checkErr(t, err)
 		if len(mods) != 1 {
 			t.Fatalf("should have had %d modified instance", 1)
 		}
@@ -896,7 +904,7 @@ func TestModifiedSince(t *testing.T) {
 		})
 		checkErr(t, err)
 
-		before := time.Now().UnixNano()
+		before := time.Now().Add(-time.Second).UnixNano()
 
 		// Create
 		newPerson := util.JSONFromInstance(Person{Name: "Alice", Age: 42})
@@ -929,9 +937,14 @@ func TestModifiedSince(t *testing.T) {
 		})
 		checkErr(t, err)
 
-		mods, err := c.ModifiedSince(before)
+		var mods []core.InstanceID
+		err = c.ReadTxn(func(txn *Txn) (err error) {
+			mods, err = txn.ModifiedSince(before)
+			return
+		})
+		checkErr(t, err)
 		if len(mods) != 2 {
-			t.Fatalf("should have had %d modified instance", 2)
+			t.Fatalf("should have had %d modified instances", 2)
 		}
 		if mods[0] != res[0] {
 			t.Fatalf("should have modfied id %s", res[0])
diff --git a/test/addr_book_suite.go b/test/addr_book_suite.go
index ea2fa1c..48a93b4 100644
--- a/test/addr_book_suite.go
+++ b/test/addr_book_suite.go
@@ -125,7 +125,7 @@ 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.Sleep(2 * time.Second)
 			// 1 seconds left.
 
 			// We still have the address.