refactor: Remove redundant txn param from fetcher start (sourcenetwor…

…k#1635) ## Relevant issue(s) Resolves sourcenetwork#1623 ## Description Removes the redundant txn param from fetcher start as it is always the same value as fetcher.Init
shahzadlone · Jul 17, 2023 · 5edc205 · 5edc205
1 parent ae01b4c
commit 5edc205
Show file tree

Hide file tree

Showing 10 changed files with 53 additions and 64 deletions.
diff --git a/db/collection_get.go b/db/collection_get.go
@@ -63,7 +63,7 @@ func (c *collection) get(
 	// construct target key for DocKey
 	targetKey := base.MakeDocKey(*desc, key.DocKey)
 	// run the doc fetcher
-	err = df.Start(ctx, txn, core.NewSpans(core.NewSpan(targetKey, targetKey.PrefixEnd())))
+	err = df.Start(ctx, core.NewSpans(core.NewSpan(targetKey, targetKey.PrefixEnd())))
 	if err != nil {
 		_ = df.Close()
 		return nil, err

diff --git a/db/collection_index.go b/db/collection_index.go
@@ -259,7 +259,7 @@ func (c *collection) iterateAllDocs(
 	start := base.MakeCollectionKey(c.desc)
 	spans := core.NewSpans(core.NewSpan(start, start.PrefixEnd()))
 
-	err = df.Start(ctx, txn, spans)
+	err = df.Start(ctx, spans)
 	if err != nil {
 		_ = df.Close()
 		return err

diff --git a/db/fetcher/fetcher.go b/db/fetcher/fetcher.go
@@ -40,7 +40,7 @@ type Fetcher interface {
 		reverse bool,
 		showDeleted bool,
 	) error
-	Start(ctx context.Context, txn datastore.Txn, spans core.Spans) error
+	Start(ctx context.Context, spans core.Spans) error
 	FetchNext(ctx context.Context) (EncodedDocument, error)
 	FetchNextDecoded(ctx context.Context) (*client.Document, error)
 	FetchNextDoc(ctx context.Context, mapping *core.DocumentMapping) ([]byte, core.Doc, error)
@@ -118,6 +118,7 @@ func (df *DocumentFetcher) Init(
 	reverse bool,
 	showDeleted bool,
 ) error {
+	df.txn = txn
 	if col.Schema.IsEmpty() {
 		return client.NewErrUninitializeProperty("DocumentFetcher", "Schema")
 	}
@@ -130,6 +131,7 @@ func (df *DocumentFetcher) Init(
 	if showDeleted {
 		if df.deletedDocFetcher == nil {
 			df.deletedDocFetcher = new(DocumentFetcher)
+			df.deletedDocFetcher.txn = txn
 		}
 		return df.deletedDocFetcher.init(col, fields, filter, docmapper, reverse)
 	}
@@ -200,21 +202,21 @@ func (df *DocumentFetcher) init(
 	return nil
 }
 
-func (df *DocumentFetcher) Start(ctx context.Context, txn datastore.Txn, spans core.Spans) error {
-	err := df.start(ctx, txn, spans, false)
+func (df *DocumentFetcher) Start(ctx context.Context, spans core.Spans) error {
+	err := df.start(ctx, spans, false)
 	if err != nil {
 		return err
 	}
 
 	if df.deletedDocFetcher != nil {
-		return df.deletedDocFetcher.start(ctx, txn, spans, true)
+		return df.deletedDocFetcher.start(ctx, spans, true)
 	}
 
 	return nil
 }
 
 // Start implements DocumentFetcher.
-func (df *DocumentFetcher) start(ctx context.Context, txn datastore.Txn, spans core.Spans, withDeleted bool) error {
+func (df *DocumentFetcher) start(ctx context.Context, spans core.Spans, withDeleted bool) error {
 	if df.col == nil {
 		return client.NewErrUninitializeProperty("DocumentFetcher", "CollectionDescription")
 	}
@@ -251,7 +253,6 @@ func (df *DocumentFetcher) start(ctx context.Context, txn datastore.Txn, spans c
 	}
 
 	df.curSpanIndex = -1
-	df.txn = txn
 
 	if df.reverse {
 		df.order = []dsq.Order{dsq.OrderByKeyDescending{}}

diff --git a/db/fetcher/mocks/Fetcher.go b/db/fetcher/mocks/Fetcher.go
diff --git a/db/fetcher/mocks/utils.go b/db/fetcher/mocks/utils.go
@@ -31,7 +31,7 @@ func NewStubbedFetcher(t *testing.T) *Fetcher {
 		mock.Anything,
 		mock.Anything,
 	).Maybe().Return(nil)
-	f.EXPECT().Start(mock.Anything, mock.Anything, mock.Anything).Maybe().Return(nil)
+	f.EXPECT().Start(mock.Anything, mock.Anything).Maybe().Return(nil)
 	f.EXPECT().FetchNext(mock.Anything).Maybe().Return(nil, nil)
 	f.EXPECT().FetchNextDoc(mock.Anything, mock.Anything).Maybe().
 		Return(NewEncodedDocument(t), core.Doc{}, nil)

diff --git a/db/fetcher/versioned.go b/db/fetcher/versioned.go
@@ -111,14 +111,29 @@ func (vf *VersionedFetcher) Init(
 	vf.col = col
 	vf.queuedCids = list.New()
 	vf.mCRDTs = make(map[uint32]crdt.MerkleCRDT)
+	vf.txn = txn
+
+	// create store
+	root := memory.NewDatastore(ctx)
+	vf.root = root
+
+	var err error
+	vf.store, err = datastore.NewTxnFrom(
+		ctx,
+		vf.root,
+		false,
+	) // were going to discard and nuke this later
+	if err != nil {
+		return err
+	}
 
 	// run the DF init, VersionedFetchers only supports the Primary (0) index
 	vf.DocumentFetcher = new(DocumentFetcher)
-	return vf.DocumentFetcher.Init(ctx, txn, col, fields, filter, docmapper, reverse, showDeleted)
+	return vf.DocumentFetcher.Init(ctx, vf.store, col, fields, filter, docmapper, reverse, showDeleted)
 }
 
 // Start serializes the correct state according to the Key and CID.
-func (vf *VersionedFetcher) Start(ctx context.Context, txn datastore.Txn, spans core.Spans) error {
+func (vf *VersionedFetcher) Start(ctx context.Context, spans core.Spans) error {
 	if vf.col == nil {
 		return client.NewErrUninitializeProperty("VersionedFetcher", "CollectionDescription")
 	}
@@ -145,29 +160,15 @@ func (vf *VersionedFetcher) Start(ctx context.Context, txn datastore.Txn, spans
 		return NewErrFailedToDecodeCIDForVFetcher(err)
 	}
 
-	vf.txn = txn
 	vf.ctx = ctx
 	vf.key = dk
 	vf.version = c
 
-	// create store
-	root := memory.NewDatastore(ctx)
-	vf.root = root
-
-	vf.store, err = datastore.NewTxnFrom(
-		ctx,
-		vf.root,
-		false,
-	) // were going to discard and nuke this later
-	if err != nil {
-		return err
-	}
-
 	if err := vf.seekTo(vf.version); err != nil {
 		return NewErrFailedToSeek(c, err)
 	}
 
-	return vf.DocumentFetcher.Start(ctx, vf.store, core.Spans{})
+	return vf.DocumentFetcher.Start(ctx, core.Spans{})
 }
 
 // Rootstore returns the rootstore of the VersionedFetcher.
@@ -196,7 +197,7 @@ func (vf *VersionedFetcher) SeekTo(ctx context.Context, c cid.Cid) error {
 		return err
 	}
 
-	return vf.DocumentFetcher.Start(ctx, vf.store, core.Spans{})
+	return vf.DocumentFetcher.Start(ctx, core.Spans{})
 }
 
 // seekTo seeks to the given CID version by stepping through the CRDT state graph from the beginning

diff --git a/db/fetcher_test.go b/db/fetcher_test.go
@@ -82,24 +82,14 @@ func TestFetcherStart(t *testing.T) {
 	df, err := newTestFetcher(ctx, txn)
 	assert.NoError(t, err)
 
-	err = df.Start(ctx, txn, core.Spans{})
+	err = df.Start(ctx, core.Spans{})
 	assert.NoError(t, err)
 }
 
 func TestFetcherStartWithoutInit(t *testing.T) {
 	ctx := context.Background()
-	db, err := newMemoryDB(ctx)
-	if err != nil {
-		t.Error(err)
-		return
-	}
-	txn, err := db.NewTxn(ctx, true)
-	if err != nil {
-		t.Error(err)
-		return
-	}
 	df := new(fetcher.DocumentFetcher)
-	err = df.Start(ctx, txn, core.Spans{})
+	err := df.Start(ctx, core.Spans{})
 	assert.Error(t, err)
 }
 
@@ -138,7 +128,7 @@ func TestFetcherGetAllPrimaryIndexEncodedDocSingle(t *testing.T) {
 	err = df.Init(ctx, txn, &desc, desc.Schema.Fields, nil, nil, false, false)
 	assert.NoError(t, err)
 
-	err = df.Start(ctx, txn, core.Spans{})
+	err = df.Start(ctx, core.Spans{})
 	assert.NoError(t, err)
 
 	encdoc, err := df.FetchNext(ctx)
@@ -183,7 +173,7 @@ func TestFetcherGetAllPrimaryIndexEncodedDocMultiple(t *testing.T) {
 	err = df.Init(ctx, txn, &desc, desc.Schema.Fields, nil, nil, false, false)
 	assert.NoError(t, err)
 
-	err = df.Start(ctx, txn, core.Spans{})
+	err = df.Start(ctx, core.Spans{})
 	assert.NoError(t, err)
 
 	encdoc, err := df.FetchNext(ctx)
@@ -221,7 +211,7 @@ func TestFetcherGetAllPrimaryIndexDecodedSingle(t *testing.T) {
 	err = df.Init(ctx, txn, &desc, desc.Schema.Fields, nil, nil, false, false)
 	assert.NoError(t, err)
 
-	err = df.Start(ctx, txn, core.Spans{})
+	err = df.Start(ctx, core.Spans{})
 	assert.NoError(t, err)
 
 	ddoc, err := df.FetchNextDecoded(ctx)
@@ -273,7 +263,7 @@ func TestFetcherGetAllPrimaryIndexDecodedMultiple(t *testing.T) {
 	err = df.Init(ctx, txn, &desc, desc.Schema.Fields, nil, nil, false, false)
 	assert.NoError(t, err)
 
-	err = df.Start(ctx, txn, core.Spans{})
+	err = df.Start(ctx, core.Spans{})
 	assert.NoError(t, err)
 
 	ddoc, err := df.FetchNextDecoded(ctx)
@@ -336,7 +326,7 @@ func TestFetcherGetOnePrimaryIndexDecoded(t *testing.T) {
 		core.NewSpan(docKey, docKey.PrefixEnd()),
 	)
 
-	err = df.Start(ctx, txn, spans)
+	err = df.Start(ctx, spans)
 	assert.NoError(t, err)
 
 	ddoc, err := df.FetchNextDecoded(ctx)

diff --git a/db/indexed_docs_test.go b/db/indexed_docs_test.go
@@ -559,8 +559,8 @@ func TestNonUniqueCreate_IfUponIndexingExistingDocsFetcherFails_ReturnError(t *t
 			Name: "Fails to start",
 			PrepareFetcher: func() fetcher.Fetcher {
 				f := fetcherMocks.NewStubbedFetcher(t)
-				f.EXPECT().Start(mock.Anything, mock.Anything, mock.Anything).Unset()
-				f.EXPECT().Start(mock.Anything, mock.Anything, mock.Anything).Return(testError)
+				f.EXPECT().Start(mock.Anything, mock.Anything).Unset()
+				f.EXPECT().Start(mock.Anything, mock.Anything).Return(testError)
 				f.EXPECT().Close().Unset()
 				f.EXPECT().Close().Return(nil)
 				return f
@@ -843,8 +843,8 @@ func TestNonUniqueUpdate_IfFetcherFails_ReturnError(t *testing.T) {
 			Name: "Fails to start",
 			PrepareFetcher: func() fetcher.Fetcher {
 				f := fetcherMocks.NewStubbedFetcher(t)
-				f.EXPECT().Start(mock.Anything, mock.Anything, mock.Anything).Unset()
-				f.EXPECT().Start(mock.Anything, mock.Anything, mock.Anything).Return(testError)
+				f.EXPECT().Start(mock.Anything, mock.Anything).Unset()
+				f.EXPECT().Start(mock.Anything, mock.Anything).Return(testError)
 				f.EXPECT().Close().Unset()
 				f.EXPECT().Close().Return(nil)
 				return f

diff --git a/lens/fetcher.go b/lens/fetcher.go
@@ -104,10 +104,8 @@ func (f *lensedFetcher) Init(
 	return f.source.Init(ctx, txn, col, innerFetcherFields, filter, docmapper, reverse, showDeleted)
 }
 
-func (f *lensedFetcher) Start(ctx context.Context, txn datastore.Txn, spans core.Spans) error {
-	f.txn = txn
-
-	return f.source.Start(ctx, txn, spans)
+func (f *lensedFetcher) Start(ctx context.Context, spans core.Spans) error {
+	return f.source.Start(ctx, spans)
 }
 
 func (f *lensedFetcher) FetchNext(ctx context.Context) (fetcher.EncodedDocument, error) {

diff --git a/planner/scan.go b/planner/scan.go
@@ -147,7 +147,7 @@ func (n *scanNode) initScan() error {
 		n.spans = core.NewSpans(core.NewSpan(start, start.PrefixEnd()))
 	}
 
-	err := n.fetcher.Start(n.p.ctx, n.p.txn, n.spans)
+	err := n.fetcher.Start(n.p.ctx, n.spans)
 	if err != nil {
 		return err
 	}