feat: Upstream BlockSTM Fork (#25483)

Co-authored-by: yihuang <[email protected]>
Co-authored-by: mmsqe <[email protected]>
Co-authored-by: mmsqe <[email protected]>
Co-authored-by: Tyler <[email protected]>

Author: 4 people

.github/workflows/test.yml

@@ -234,6 +234,11 @@ jobs:
         with:
           name: "${{ github.sha }}-store-coverage"
         continue-on-error: true
+      - uses: actions/download-artifact@v4
+        if: env.GIT_DIFF
+        with:
+          name: "${{ github.sha }}-blockstm-coverage"
+        continue-on-error: true
       - uses: actions/download-artifact@v4
         if: env.GIT_DIFF
         with:
@@ -253,7 +258,7 @@ jobs:
         if: env.GIT_DIFF
         uses: codecov/codecov-action@v5
         with:
-          files: ./00profile.out,./01profile.out,./02profile.out,./03profile.out,./integration-profile.out,./e2e-profile.out,./client/v2/coverage.out,./core/coverage.out,./depinject/coverage.out,./errors/coverage.out,./math/coverage.out,./schema/coverage.out,./collections/coverage.out,./tools/cosmovisor/coverage.out,./tools/confix/coverage.out,./store/coverage.out,./log/coverage.out,./x/tx/coverage.out,./tools/benchmark/coverage.out
+          files: ./00profile.out,./01profile.out,./02profile.out,./03profile.out,./integration-profile.out,./e2e-profile.out,./client/v2/coverage.out,./core/coverage.out,./depinject/coverage.out,./errors/coverage.out,./math/coverage.out,./schema/coverage.out,./collections/coverage.out,./tools/cosmovisor/coverage.out,./tools/confix/coverage.out,./store/coverage.out,./log/coverage.out,./x/tx/coverage.out,./tools/benchmark/coverage.out,./blockstm/coverage.out
           fail_ci_if_error: false
           verbose: true
           token: ${{ secrets.CODECOV_TOKEN }}
@@ -570,6 +575,33 @@ jobs:
         with:
           name: "${{ github.sha }}-store-coverage"
           path: ./store/coverage.out
+  test-blockstm:
+    runs-on: depot-ubuntu-22.04-4
+    steps:
+      - uses: actions/checkout@v5
+      - uses: actions/setup-go@v6
+        with:
+          go-version: "1.25"
+          check-latest: true
+          cache: true
+          cache-dependency-path: store/go.sum
+      - uses: technote-space/[email protected]
+        id: git_diff
+        with:
+          PATTERNS: |
+            blockstm/**/*.go
+            blockstm/go.mod
+            blockstm/go.sum
+      - name: tests
+        if: env.GIT_DIFF
+        run: |
+          cd blockstm
+          go test -mod=readonly -timeout 30m -coverprofile=coverage.out -covermode=atomic -coverpkg=./,./tree -tags='norace ledger test_ledger_mock' ./...
+      - uses: actions/upload-artifact@v4
+        if: env.GIT_DIFF
+        with:
+          name: "${{ github.sha }}-blockstm-coverage"
+          path: ./blockstm/coverage.out
 
   test-log:
     runs-on: depot-ubuntu-22.04-4

blockstm/README.md

@@ -0,0 +1,41 @@
+`blockstm` implements the [block-stm algorithm](https://arxiv.org/abs/2203.06871), it follows the paper pseudocode pretty closely.
+
+The main API is a simple function call `ExecuteBlock`:
+
+```golang
+type ExecuteFn func(TxnIndex, MultiStore)
+func ExecuteBlock(
+ ctx context.Context,           // context for cancellation
+ blockSize int,                 // the number of the transactions to be executed
+ stores []storetypes.StoreKey,  // the list of store keys to support
+ storage MultiStore,            // the parent storage, after all transactions are executed, the whole change sets are written into parent storage at once
+ executors int,                 // how many concurrent executors to spawn
+ executeFn ExecuteFn,           // callback function to actually execute a transaction with a wrapped `MultiStore`.
+) error
+```
+
+The main deviations from the paper are:
+
+### Optimisation
+
+We applied the optimization described in section 4 of the paper:
+
+```
+Block-STM calls add_dependency from the VM itself, and can thus re-read and continue execution when false is returned.
+```
+
+When the VM execution reads an `ESTIMATE` mark, it'll hang on a `CondVar`, so it can resume execution after the dependency is resolved,
+much more efficient than abortion and rerun.
+
+### Support Deletion, Iteration, and MultiStore
+
+These features are necessary for integration with cosmos-sdk.
+
+The multi-version data structure is implemented with nested btree for easier iteration support,
+the `WriteSet` is also implemented with a btree, and it takes advantage of ordered property to optimize some logic.
+
+The internal data structures are also adapted with multiple stores in mind.
+
+### Attribution
+
+This package was originally authored in [go-block-stm](https://github.com/crypto-org-chain/go-block-stm). We have brought the full source tree into the SDK so that we can natively incorporate the library and required changes into the SDK. Over time we expect to incoporate optimizations and deviations from the upstream implementation.

blockstm/bench_test.go

@@ -0,0 +1,54 @@
+package blockstm
+
+import (
+	"context"
+	"strconv"
+	"testing"
+
+	"github.com/test-go/testify/require"
+
+	storetypes "cosmossdk.io/store/types"
+)
+
+func BenchmarkBlockSTM(b *testing.B) {
+	stores := map[storetypes.StoreKey]int{StoreKeyAuth: 0, StoreKeyBank: 1}
+	for i := 0; i < 26; i++ {
+		key := storetypes.NewKVStoreKey(strconv.FormatInt(int64(i), 10))
+		stores[key] = i + 2
+	}
+	storage := NewMultiMemDB(stores)
+	testCases := []struct {
+		name  string
+		block *MockBlock
+	}{
+		{"random-10000/100", testBlock(10000, 100)},
+		{"no-conflict-10000", noConflictBlock(10000)},
+		{"worst-case-10000", worstCaseBlock(10000)},
+		{"iterate-10000/100", iterateBlock(10000, 100)},
+	}
+	for _, tc := range testCases {
+		b.Run(tc.name+"-sequential", func(b *testing.B) {
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				runSequential(storage, tc.block)
+			}
+		})
+		for _, worker := range []int{1, 5, 10, 15, 20} {
+			b.Run(tc.name+"-worker-"+strconv.Itoa(worker), func(b *testing.B) {
+				b.ResetTimer()
+				for i := 0; i < b.N; i++ {
+					require.NoError(
+						b,
+						ExecuteBlock(context.Background(), tc.block.Size(), stores, storage, worker, tc.block.ExecuteTx),
+					)
+				}
+			})
+		}
+	}
+}
+
+func runSequential(storage MultiStore, block *MockBlock) {
+	for i, tx := range block.Txs {
+		block.Results[i] = tx(storage)
+	}
+}

blockstm/condvar.go

@@ -0,0 +1,30 @@
+package blockstm
+
+import "sync"
+
+type Condvar struct {
+	sync.Mutex
+	notified bool
+	cond     sync.Cond
+}
+
+func NewCondvar() *Condvar {
+	c := &Condvar{}
+	c.cond = *sync.NewCond(c)
+	return c
+}
+
+func (cv *Condvar) Wait() {
+	cv.Lock()
+	for !cv.notified {
+		cv.cond.Wait()
+	}
+	cv.Unlock()
+}
+
+func (cv *Condvar) Notify() {
+	cv.Lock()
+	cv.notified = true
+	cv.Unlock()
+	cv.cond.Signal()
+}

blockstm/executor.go

@@ -0,0 +1,89 @@
+package blockstm
+
+import (
+	"context"
+	"fmt"
+)
+
+// Executor fields are not mutated during execution.
+type Executor struct {
+	ctx        context.Context // context for cancellation
+	scheduler  *Scheduler      // scheduler for task management
+	txExecutor TxExecutor      // callback to actually execute a transaction
+	mvMemory   *MVMemory       // multi-version memory for the executor
+
+	// index of the executor, used for debugging output
+	i int
+}
+
+func NewExecutor(
+	ctx context.Context,
+	scheduler *Scheduler,
+	txExecutor TxExecutor,
+	mvMemory *MVMemory,
+	i int,
+) *Executor {
+	return &Executor{
+		ctx:        ctx,
+		scheduler:  scheduler,
+		txExecutor: txExecutor,
+		mvMemory:   mvMemory,
+		i:          i,
+	}
+}
+
+// Run executes all tasks until completion
+// Invariant `num_active_tasks`:
+//   - `NextTask` increases it if returns a valid task.
+//   - `TryExecute` and `NeedsReexecution` don't change it if it returns a new valid task to run,
+//     otherwise it decreases it.
+func (e *Executor) Run() error {
+	var kind TaskKind
+	version := InvalidTxnVersion
+	for !e.scheduler.Done() {
+		if !version.Valid() {
+			// check for cancellation
+			select {
+			case <-e.ctx.Done():
+				return nil
+			default:
+			}
+
+			version, kind = e.scheduler.NextTask()
+			continue
+		}
+
+		switch kind {
+		case TaskKindExecution:
+			version, kind = e.TryExecute(version)
+		case TaskKindValidation:
+			version, kind = e.NeedsReexecution(version)
+		default:
+			return fmt.Errorf("unknown task kind %v", kind)
+		}
+	}
+	return nil
+}
+
+func (e *Executor) TryExecute(version TxnVersion) (TxnVersion, TaskKind) {
+	e.scheduler.executedTxns.Add(1)
+	view := e.execute(version.Index)
+	wroteNewLocation := e.mvMemory.Record(version, view)
+	return e.scheduler.FinishExecution(version, wroteNewLocation)
+}
+
+func (e *Executor) NeedsReexecution(version TxnVersion) (TxnVersion, TaskKind) {
+	e.scheduler.validatedTxns.Add(1)
+	valid := e.mvMemory.ValidateReadSet(version.Index)
+	aborted := !valid && e.scheduler.TryValidationAbort(version)
+	if aborted {
+		e.mvMemory.ConvertWritesToEstimates(version.Index)
+	}
+	return e.scheduler.FinishValidation(version.Index, aborted)
+}
+
+func (e *Executor) execute(txn TxnIndex) *MultiMVMemoryView {
+	view := e.mvMemory.View(txn)
+	e.txExecutor(txn, view)
+	return view
+}