chore(embedded/store): allow skipping precommitted transactions during open

Signed-off-by: Stefano Scafiti <[email protected]>

Author: ostafen

embedded/store/immustore.go

@@ -533,43 +533,45 @@ func OpenWith(path string, vLogs []appendable.Appendable, txLog, cLog appendable
 	precommittedAlh := committedAlh
 	precommittedTxLogSize := committedTxLogSize
 
-	// read pre-committed txs from txLog and insert into cLogBuf to continue with the commit process
-	// txLog may be partially written, precommitted transactions loading is terminated if an inconsistency is found
-	txReader := appendable.NewReaderFrom(txLog, precommittedTxLogSize, multiapp.DefaultReadBufferSize)
+	if !opts.SkipPrecommittedTransactions {
+		// read pre-committed txs from txLog and insert into cLogBuf to continue with the commit process
+		// txLog may be partially written, precommitted transactions loading is terminated if an inconsistency is found
+		txReader := appendable.NewReaderFrom(txLog, precommittedTxLogSize, multiapp.DefaultReadBufferSize)
 
-	tx, _ := txPool.Alloc()
+		tx, _ := txPool.Alloc()
 
-	for {
-		err = tx.readFrom(txReader, false)
-		if errors.Is(err, io.EOF) {
-			break
-		}
-		if err != nil {
-			opts.logger.Infof("%v: discarding pre-committed transaction: %d", err, precommittedTxID+1)
-			break
-		}
+		for {
+			err = tx.readFrom(txReader, false)
+			if errors.Is(err, io.EOF) {
+				break
+			}
+			if err != nil {
+				opts.logger.Infof("%v: discarding pre-committed transaction: %d", err, precommittedTxID+1)
+				break
+			}
 
-		if tx.header.ID != precommittedTxID+1 || tx.header.PrevAlh != precommittedAlh {
-			opts.logger.Infof("%v: discarding pre-committed transaction: %d", ErrCorruptedData, precommittedTxID+1)
-			break
-		}
+			if tx.header.ID != precommittedTxID+1 || tx.header.PrevAlh != precommittedAlh {
+				opts.logger.Infof("%v: discarding pre-committed transaction: %d", ErrCorruptedData, precommittedTxID+1)
+				break
+			}
 
-		precommittedTxID++
-		precommittedAlh = tx.header.Alh()
+			precommittedTxID++
+			precommittedAlh = tx.header.Alh()
 
-		txSize := int(txReader.ReadCount() - (precommittedTxLogSize - committedTxLogSize))
+			txSize := int(txReader.ReadCount() - (precommittedTxLogSize - committedTxLogSize))
 
-		err = cLogBuf.put(precommittedTxID, precommittedAlh, precommittedTxLogSize, txSize)
-		if err != nil {
-			txPool.Release(tx)
-			return nil, fmt.Errorf("%v: while loading pre-committed transaction: %v", err, precommittedTxID+1)
+			err = cLogBuf.put(precommittedTxID, precommittedAlh, precommittedTxLogSize, txSize)
+			if err != nil {
+				txPool.Release(tx)
+				return nil, fmt.Errorf("%v: while loading pre-committed transaction: %v", err, precommittedTxID+1)
+			}
+
+			precommittedTxLogSize += int64(txSize)
 		}
 
-		precommittedTxLogSize += int64(txSize)
+		txPool.Release(tx)
 	}
 
-	txPool.Release(tx)
-
 	vLogsMap := make(map[byte]*refVLog, len(vLogs))
 	vLogUnlockedList := list.New()
 

embedded/store/options.go

@@ -147,6 +147,8 @@ type Options struct {
 	CompressionLevel  int
 	EmbeddedValues    bool
 	PreallocFiles     bool
+	// Skip processing of transactions that were precommitted before opening
+	SkipPrecommittedTransactions bool
 
 	// options below affect indexing
 	IndexOpts *IndexOptions
@@ -247,17 +249,17 @@ func DefaultOptions() *Options {
 		WriteTxHeaderVersion: DefaultWriteTxHeaderVersion,
 
 		// options below are only set during initialization and stored as metadata
-		MaxTxEntries:      DefaultMaxTxEntries,
-		MaxKeyLen:         DefaultMaxKeyLen,
-		MaxValueLen:       DefaultMaxValueLen,
-		FileSize:          DefaultFileSize,
-		CompressionFormat: DefaultCompressionFormat,
-		CompressionLevel:  DefaultCompressionLevel,
-		EmbeddedValues:    DefaultEmbeddedValues,
-		PreallocFiles:     DefaultPreallocFiles,
-
-		IndexOpts: DefaultIndexOptions(),
-		AHTOpts:   DefaultAHTOptions(),
+		MaxTxEntries:                 DefaultMaxTxEntries,
+		MaxKeyLen:                    DefaultMaxKeyLen,
+		MaxValueLen:                  DefaultMaxValueLen,
+		FileSize:                     DefaultFileSize,
+		CompressionFormat:            DefaultCompressionFormat,
+		CompressionLevel:             DefaultCompressionLevel,
+		EmbeddedValues:               DefaultEmbeddedValues,
+		PreallocFiles:                DefaultPreallocFiles,
+		SkipPrecommittedTransactions: false,
+		IndexOpts:                    DefaultIndexOptions(),
+		AHTOpts:                      DefaultAHTOptions(),
 	}
 }
 
@@ -620,6 +622,11 @@ func (opts *Options) WithAHTOptions(ahtOptions *AHTOptions) *Options {
 	return opts
 }
 
+func (opts *Options) WithSkipPrecommittedTransactions(skip bool) *Options {
+	opts.SkipPrecommittedTransactions = true
+	return opts
+}
+
 // IndexOptions
 
 func (opts *IndexOptions) WithCacheSize(cacheSize int) *IndexOptions {