fix: Add version check in basicTxn.Query

datastore/memory/txn.go

@@ -179,13 +179,17 @@ func (t *basicTxn) Query(ctx context.Context, q dsq.Query) (dsq.Results, error)
 	iter := t.ds.values.Iter()
 	iterOps := t.ops.Iter()
 	iterOpsHasValue := iterOps.Next()
+	dsVersion := t.getDSVersion()
 	// iterate over the underlying store and ensure that ops with keys smaller than or equal to
 	// the key of the underlying store are added with priority.
 	for iter.Next() {
 		// fast forward to last inserted version
 		item := iter.Item()
+		if item.version > dsVersion {
+			continue
+		}
 		for iter.Next() {
-			if item.key == iter.Item().key {
+			if item.key == iter.Item().key && iter.Item().version <= dsVersion {
 				item = iter.Item()
 				continue
 			}

datastore/memory/txn_test.go

@@ -661,7 +661,7 @@ func TestTxnQueryWithOnlyOneOperation(t *testing.T) {
 	tx, err := s.NewTransaction(ctx, false)
 	require.NoError(t, err)
 
-	err = s.Put(ctx, testKey4, testValue4)
+	err = tx.Put(ctx, testKey4, testValue4)
 	require.NoError(t, err)
 
 	results, err := tx.Query(ctx, dsq.Query{})

datastore/txn_test.go

@@ -16,6 +16,7 @@ import (
 	"testing"
 
 	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/query"
 	badger "github.com/sourcenetwork/badger/v4"
 	"github.com/stretchr/testify/require"
 
@@ -458,3 +459,94 @@ func TestBadgerFileStoreTxn_TwoTransactionsWithHasPutConflict_ShouldErrorWithCon
 	err = txn1.Commit(ctx)
 	require.ErrorIs(t, err, badger.ErrConflict)
 }
+
+func TestMemoryStoreTxn_TwoTransactionsWithQueryAndPut_ShouldOmmitNewPut(t *testing.T) {
+	ctx := context.Background()
+	rootstore := memory.NewDatastore(ctx)
+
+	rootstore.Put(ctx, ds.NewKey("key"), []byte("value"))
+
+	txn1, err := rootstore.NewTransaction(ctx, false)
+	require.NoError(t, err)
+
+	txn2, err := rootstore.NewTransaction(ctx, false)
+	require.NoError(t, err)
+
+	err = txn2.Put(ctx, ds.NewKey("other-key"), []byte("other-value"))
+	require.NoError(t, err)
+
+	err = txn2.Commit(ctx)
+	require.NoError(t, err)
+
+	qResults, err := txn1.Query(ctx, query.Query{})
+	require.NoError(t, err)
+
+	docs := [][]byte{}
+	for r := range qResults.Next() {
+		docs = append(docs, r.Entry.Value)
+	}
+	require.Equal(t, [][]byte{[]byte("value")}, docs)
+	txn1.Discard(ctx)
+}
+
+func TestBadgerMemoryStoreTxn_TwoTransactionsWithQueryAndPut_ShouldOmmitNewPut(t *testing.T) {
+	ctx := context.Background()
+	opts := badgerds.Options{Options: badger.DefaultOptions("").WithInMemory(true)}
+	rootstore, err := badgerds.NewDatastore("", &opts)
+	require.NoError(t, err)
+
+	rootstore.Put(ctx, ds.NewKey("key"), []byte("value"))
+
+	txn1, err := rootstore.NewTransaction(ctx, false)
+	require.NoError(t, err)
+
+	txn2, err := rootstore.NewTransaction(ctx, false)
+	require.NoError(t, err)
+
+	err = txn2.Put(ctx, ds.NewKey("other-key"), []byte("other-value"))
+	require.NoError(t, err)
+
+	err = txn2.Commit(ctx)
+	require.NoError(t, err)
+
+	qResults, err := txn1.Query(ctx, query.Query{})
+	require.NoError(t, err)
+
+	docs := [][]byte{}
+	for r := range qResults.Next() {
+		docs = append(docs, r.Entry.Value)
+	}
+	require.Equal(t, [][]byte{[]byte("value")}, docs)
+	txn1.Discard(ctx)
+}
+
+func TestBadgerFileStoreTxn_TwoTransactionsWithQueryAndPut_ShouldOmmitNewPut(t *testing.T) {
+	ctx := context.Background()
+	opts := badgerds.Options{Options: badger.DefaultOptions("")}
+	rootstore, err := badgerds.NewDatastore(t.TempDir(), &opts)
+	require.NoError(t, err)
+
+	rootstore.Put(ctx, ds.NewKey("key"), []byte("value"))
+
+	txn1, err := rootstore.NewTransaction(ctx, false)
+	require.NoError(t, err)
+
+	txn2, err := rootstore.NewTransaction(ctx, false)
+	require.NoError(t, err)
+
+	err = txn2.Put(ctx, ds.NewKey("other-key"), []byte("other-value"))
+	require.NoError(t, err)
+
+	err = txn2.Commit(ctx)
+	require.NoError(t, err)
+
+	qResults, err := txn1.Query(ctx, query.Query{})
+	require.NoError(t, err)
+
+	docs := [][]byte{}
+	for r := range qResults.Next() {
+		docs = append(docs, r.Entry.Value)
+	}
+	require.Equal(t, [][]byte{[]byte("value")}, docs)
+	txn1.Discard(ctx)
+}