ledger: move MakeCatchpointReader back to the Reader interface (#5583)

algorand · algojohnlee · Aug 4, 2023 · May 31, 2023 · May 31, 2023 · Jun 1, 2023
commit 197bfb6a08b29582672be2c07e2fa495036185cd
diff --git a/ledger/catchpointtracker.go b/ledger/catchpointtracker.go
@@ -1151,7 +1151,7 @@ func (ct *catchpointTracker) generateCatchpointData(ctx context.Context, account
 
 	start := time.Now()
 	ledgerGeneratecatchpointCount.Inc(nil)
-	err = ct.dbs.TransactionContext(ctx, func(dbCtx context.Context, tx trackerdb.TransactionScope) (err error) {
+	err = ct.dbs.SnapshotContext(ctx, func(dbCtx context.Context, tx trackerdb.SnapshotScope) (err error) {
 		catchpointWriter, err = makeCatchpointWriter(dbCtx, catchpointDataFilePath, tx, ResourcesPerCatchpointFileChunk)
 		if err != nil {
 			return
@@ -1341,7 +1341,7 @@ func (ct *catchpointTracker) GetCatchpointStream(round basics.Round) (ReadCloseS
 	ledgerGetcatchpointCount.Inc(nil)
 	// TODO: we need to generalize this, check @cce PoC PR, he has something
 	//       somewhat broken for some KVs..
-	err := ct.dbs.Transaction(func(ctx context.Context, tx trackerdb.TransactionScope) (err error) {
+	err := ct.dbs.Snapshot(func(ctx context.Context, tx trackerdb.SnapshotScope) (err error) {
 		cr, err := tx.MakeCatchpointReader()
 		if err != nil {
 			return err

diff --git a/ledger/catchpointwriter.go b/ledger/catchpointwriter.go
@@ -52,7 +52,7 @@ const (
 // has the option of throttling the CPU utilization in between the calls.
 type catchpointWriter struct {
 	ctx                  context.Context
-	tx                   trackerdb.TransactionScope
+	tx                   trackerdb.SnapshotScope
 	filePath             string
 	totalAccounts        uint64
 	totalKVs             uint64
@@ -106,7 +106,7 @@ func (data catchpointStateProofVerificationContext) ToBeHashed() (protocol.HashI
 	return protocol.StateProofVerCtx, protocol.Encode(&data)
 }
 
-func makeCatchpointWriter(ctx context.Context, filePath string, tx trackerdb.TransactionScope, maxResourcesPerChunk int) (*catchpointWriter, error) {
+func makeCatchpointWriter(ctx context.Context, filePath string, tx trackerdb.SnapshotScope, maxResourcesPerChunk int) (*catchpointWriter, error) {
 	aw, err := tx.MakeAccountsReader()
 	if err != nil {
 		return nil, err

diff --git a/ledger/store/trackerdb/sqlitedriver/catchpoint.go b/ledger/store/trackerdb/sqlitedriver/catchpoint.go
@@ -52,6 +52,10 @@ func NewCatchpointSQLReaderWriter(e db.Executable) *catchpointReaderWriter {
 	}
 }
 
+func makeCatchpointReader(e db.Queryable) trackerdb.CatchpointReader {
+	return &catchpointReader{q: e}
+}
+
 func (cr *catchpointReader) GetCatchpoint(ctx context.Context, round basics.Round) (fileName string, catchpoint string, fileSize int64, err error) {
 	err = cr.q.QueryRowContext(ctx, "SELECT filename, catchpoint, filesize FROM storedcatchpoints WHERE round=?", int64(round)).Scan(&fileName, &catchpoint, &fileSize)
 	return

diff --git a/ledger/store/trackerdb/sqlitedriver/sqlitedriver.go b/ledger/store/trackerdb/sqlitedriver/sqlitedriver.go
@@ -85,7 +85,7 @@ func (s *trackerSQLStore) Snapshot(fn trackerdb.SnapshotFn) (err error) {
 
 func (s *trackerSQLStore) SnapshotContext(ctx context.Context, fn trackerdb.SnapshotFn) (err error) {
 	return s.pair.Rdb.AtomicContext(ctx, func(ctx context.Context, tx *sql.Tx) error {
-		return fn(ctx, sqlSnapshotScope{tx, &sqlReader{tx}})
+		return fn(ctx, &sqlSnapshotScope{tx, &sqlReader{tx}})
 	})
 }
 
@@ -183,6 +183,21 @@ func (r *sqlReader) MakeCatchpointPendingHashesIterator(hashCount int) trackerdb
 	return MakeCatchpointPendingHashesIterator(hashCount, r.q)
 }
 
+// MakeCatchpointReader implements trackerdb.Reader
+func (r *sqlReader) MakeCatchpointReader() (trackerdb.CatchpointReader, error) {
+	return makeCatchpointReader(r.q), nil
+}
+
+// MakeEncodedAccoutsBatchIter implements trackerdb.Reader
+func (r *sqlReader) MakeEncodedAccoutsBatchIter() trackerdb.EncodedAccountsBatchIter {
+	return MakeEncodedAccoutsBatchIter(r.q)
+}
+
+// MakeKVsIter implements trackerdb.Reader
+func (r *sqlReader) MakeKVsIter(ctx context.Context) (trackerdb.KVsIter, error) {
+	return MakeKVsIter(ctx, r.q)
+}
+
 type sqlWriter struct {
 	e db.Executable
 }
@@ -236,11 +251,6 @@ type sqlCatchpoint struct {
 	e db.Executable
 }
 
-// MakeCatchpointReader implements trackerdb.Catchpoint
-func (c *sqlCatchpoint) MakeCatchpointReader() (trackerdb.CatchpointReader, error) {
-	return NewCatchpointSQLReaderWriter(c.e), nil
-}
-
 // MakeCatchpointReaderWriter implements trackerdb.Catchpoint
 func (c *sqlCatchpoint) MakeCatchpointReaderWriter() (trackerdb.CatchpointReaderWriter, error) {
 	return NewCatchpointSQLReaderWriter(c.e), nil
@@ -251,16 +261,6 @@ func (c *sqlCatchpoint) MakeCatchpointWriter() (trackerdb.CatchpointWriter, erro
 	return NewCatchpointSQLReaderWriter(c.e), nil
 }
 
-// MakeEncodedAccoutsBatchIter implements trackerdb.Catchpoint
-func (c *sqlCatchpoint) MakeEncodedAccoutsBatchIter() trackerdb.EncodedAccountsBatchIter {
-	return MakeEncodedAccoutsBatchIter(c.e)
-}
-
-// MakeKVsIter implements trackerdb.Catchpoint
-func (c *sqlCatchpoint) MakeKVsIter(ctx context.Context) (trackerdb.KVsIter, error) {
-	return MakeKVsIter(ctx, c.e)
-}
-
 // MakeMerkleCommitter implements trackerdb.Catchpoint
 func (c *sqlCatchpoint) MakeMerkleCommitter(staging bool) (trackerdb.MerkleCommitter, error) {
 	return MakeMerkleCommitter(c.e, staging)
@@ -302,7 +302,11 @@ type sqlSnapshotScope struct {
 	trackerdb.Reader
 }
 
-func (ss sqlSnapshotScope) Close() error {
+func (ss *sqlSnapshotScope) ResetTransactionWarnDeadline(ctx context.Context, deadline time.Time) (prevDeadline time.Time, err error) {
+	return db.ResetTransactionWarnDeadline(ctx, ss.tx, deadline)
+}
+
+func (ss *sqlSnapshotScope) Close() error {
 	return ss.tx.Rollback()
 }
 

diff --git a/ledger/store/trackerdb/store.go b/ledger/store/trackerdb/store.go
@@ -59,6 +59,10 @@ type Reader interface {
 	// catchpoint
 	// Note: BuildMerkleTrie() needs this on the reader handle in sqlite to not get locked by write txns
 	MakeCatchpointPendingHashesIterator(hashCount int) CatchpointPendingHashesIter
+	// Note: Catchpoint tracker needs this on the reader handle in sqlite to not get locked by write txns
+	MakeCatchpointReader() (CatchpointReader, error)
+	MakeEncodedAccoutsBatchIter() EncodedAccountsBatchIter
+	MakeKVsIter(ctx context.Context) (KVsIter, error)
 }
 
 // Writer is the interface for the trackerdb write operations.
@@ -82,10 +86,7 @@ type Writer interface {
 //	we should split these two sets of methods into two separate interfaces
 type Catchpoint interface {
 	// reader
-	MakeCatchpointReader() (CatchpointReader, error)
 	MakeOrderedAccountsIter(accountCount int) OrderedAccountsIter
-	MakeKVsIter(ctx context.Context) (KVsIter, error)
-	MakeEncodedAccoutsBatchIter() EncodedAccountsBatchIter
 	// writer
 	MakeCatchpointWriter() (CatchpointWriter, error)
 	// reader/writer
@@ -122,6 +123,7 @@ type Batch interface {
 // SnapshotScope is an atomic read-only scope to the store.
 type SnapshotScope interface {
 	Reader
+	ResetTransactionWarnDeadline(ctx context.Context, deadline time.Time) (prevDeadline time.Time, err error)
 }
 
 // Snapshot is an atomic read-only accecssor to the store.