sync: apply entire query while locked

Author: Stebalien

query/query.go

@@ -258,6 +258,23 @@ func ResultsWithEntries(q Query, res []Entry) Results {
 	})
 }
 
+// ResultsWithResults returns a Results object from a list of Results and a
+// close error.
+func ResultsWithResults(q Query, res []Result, closeErr error) Results {
+	i := 0
+	return ResultsFromIterator(q, Iterator{
+		Next: func() (Result, bool) {
+			if i >= len(res) {
+				return Result{}, false
+			}
+			next := res[i]
+			i++
+			return next, true
+		},
+		Close: func() error {
+			return closeErr
+		},
+	})
 }
 
 func ResultsReplaceQuery(r Results, q Query) Results {

sync/sync.go

@@ -61,11 +61,26 @@ func (d *MutexDatastore) Delete(key ds.Key) (err error) {
 	return d.child.Delete(key)
 }
 
-// KeyList implements Datastore.KeyList
+// Query implements Datastore.Query
 func (d *MutexDatastore) Query(q dsq.Query) (dsq.Results, error) {
 	d.RLock()
 	defer d.RUnlock()
-	return d.child.Query(q)
+	results, err := d.child.Query(q)
+	if err != nil {
+		return nil, err
+	}
+
+	// Apply the entire query while locked. Non-sync datastores may not
+	// allow concurrent queries.
+	var res []dsq.Result
+	for {
+		next, ok := results.NextSync()
+		if !ok {
+			break
+		}
+		res = append(res, next)
+	}
+	return dsq.ResultsWithResults(q, res, results.Close()), nil
 }
 
 func (d *MutexDatastore) Batch() (ds.Batch, error) {