Skip to content

Commit 824da79

Browse files
authored
refactor(sequencers): implement batch position persistance (#2908)
<!-- Please read and fill out this form before submitting your PR. Please make sure you have reviewed our contributors guide before submitting your first PR. NOTE: PR titles should follow semantic commits: https://www.conventionalcommits.org/en/v1.0.0/ --> ## Overview ref #2906 Implement da epoch position persistance in case of crash. During transactions smoothing, if we are stopping mid epoch, while a smoothing is happening, restarting will refetch the epoch, and restart executing from the last checkpoint. This allows as well all base sequencers to have deterministic blocks. As restarting at different times would lead the different txs being reexecuted (it was the same issue in single sequencers, but non determinism in sequencing does not matter). This PR fixes it for both anyway. docs are obviously ai made haha. <!-- Please provide an explanation of the PR, including the appropriate context, background, goal, and rationale. If there is an issue with this information, please provide a tl;dr and link the issue. Ex: Closes #<issue number> -->
1 parent 79310d0 commit 824da79

18 files changed

Lines changed: 1546 additions & 836 deletions

File tree

apps/evm/cmd/run.go

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -169,7 +169,10 @@ func createSequencer(
169169
return nil, fmt.Errorf("based sequencer mode requires aggregator mode to be enabled")
170170
}
171171

172-
basedSeq := based.NewBasedSequencer(fiRetriever, da, nodeConfig, genesis, logger)
172+
basedSeq, err := based.NewBasedSequencer(ctx, fiRetriever, datastore, genesis, logger)
173+
if err != nil {
174+
return nil, fmt.Errorf("failed to create based sequencer: %w", err)
175+
}
173176

174177
logger.Info().
175178
Str("forced_inclusion_namespace", nodeConfig.DA.GetForcedInclusionNamespace()).

apps/grpc/cmd/run.go

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -131,7 +131,10 @@ func createSequencer(
131131
return nil, fmt.Errorf("based sequencer mode requires aggregator mode to be enabled")
132132
}
133133

134-
basedSeq := based.NewBasedSequencer(fiRetriever, da, nodeConfig, genesis, logger)
134+
basedSeq, err := based.NewBasedSequencer(ctx, fiRetriever, datastore, genesis, logger)
135+
if err != nil {
136+
return nil, fmt.Errorf("failed to create based sequencer: %w", err)
137+
}
135138

136139
logger.Info().
137140
Str("forced_inclusion_namespace", nodeConfig.DA.GetForcedInclusionNamespace()).

apps/testapp/cmd/run.go

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -131,7 +131,10 @@ func createSequencer(
131131
return nil, fmt.Errorf("based sequencer mode requires aggregator mode to be enabled")
132132
}
133133

134-
basedSeq := based.NewBasedSequencer(fiRetriever, da, nodeConfig, genesis, logger)
134+
basedSeq, err := based.NewBasedSequencer(ctx, fiRetriever, datastore, genesis, logger)
135+
if err != nil {
136+
return nil, fmt.Errorf("failed to create based sequencer: %w", err)
137+
}
135138

136139
logger.Info().
137140
Str("forced_inclusion_namespace", nodeConfig.DA.GetForcedInclusionNamespace()).

pkg/blob/README.md

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,14 +3,17 @@
33
This package is a **trimmed copy** of code from `celestia-node` to stay JSON-compatible with the blob RPC without importing the full Cosmos/Celestia dependency set.
44

55
## Upstream source
6+
67
- `blob.go` comes from `celestia-node/blob/blob.go` @ tag `v0.28.4` (release v0.28.4), with unused pieces removed (blob v1, proof helpers, share length calc, appconsts dependency, etc.).
78
- `submit_options.go` mirrors the exported JSON fields of `celestia-node/state/tx_config.go` @ the same tag, leaving out functional options, defaults, and Cosmos keyring helpers.
89

910
## Why copy instead of import?
11+
1012
- Avoids pulling Cosmos SDK / celestia-app dependencies into ev-node for the small surface we need (blob JSON and commitment for v0).
1113
- Keeps binary size and module graph smaller while remaining wire-compatible with celestia-node's blob service.
1214

1315
## Keeping it in sync
16+
1417
- When celestia-node changes blob JSON or tx config fields, update this package manually:
1518
1. `diff -u pkg/blob/blob.go ../Celestia/celestia-node/blob/blob.go`
1619
2. `diff -u pkg/blob/submit_options.go ../Celestia/celestia-node/state/tx_config.go`

proto/evnode/v1/state.proto

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,3 +19,11 @@ message State {
1919

2020
reserved 7;
2121
}
22+
23+
// SequencerDACheckpoint tracks the position in the DA where transactions were last processed
24+
message SequencerDACheckpoint {
25+
// DA block height being processed
26+
uint64 da_height = 1;
27+
// Index of the next transaction to process within the DA block's forced inclusion batch
28+
uint64 tx_index = 2;
29+
}

sequencers/based/README.md

Lines changed: 204 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,204 @@
1+
# Based Sequencer
2+
3+
## Overview
4+
5+
The Based Sequencer is a sequencer implementation that exclusively retrieves transactions from the Data Availability (DA) layer via the forced inclusion mechanism. Unlike other sequencer types, it does not accept transactions from a mempool or reaper - it treats the DA layer as a transaction queue.
6+
7+
This design ensures that all transactions are force-included from DA, making the sequencer completely "based" on the DA layer's transaction ordering.
8+
9+
## Architecture
10+
11+
### Core Components
12+
13+
1. **ForcedInclusionRetriever**: Fetches transactions from DA at epoch boundaries
14+
2. **CheckpointStore**: Persists processing position to enable crash recovery
15+
3. **BasedSequencer**: Orchestrates transaction retrieval and batch creation
16+
17+
### Key Interfaces
18+
19+
The Based Sequencer implements the `Sequencer` interface from `core/sequencer.go`:
20+
21+
- `SubmitBatchTxs()` - No-op for based sequencer (transactions are not accepted)
22+
- `GetNextBatch()` - Retrieves the next batch from DA via forced inclusion
23+
- `VerifyBatch()` - Always returns true (all transactions come from DA)
24+
25+
## Epoch-Based Transaction Retrieval
26+
27+
### How Epochs Work
28+
29+
Transactions are retrieved from DA in **epochs**, not individual DA blocks. An epoch is a range of DA blocks defined by `DAEpochForcedInclusion` in the genesis configuration.
30+
31+
**Example**: If `DAStartHeight = 100` and `DAEpochForcedInclusion = 10`:
32+
33+
- Epoch 1: DA heights 100-109
34+
- Epoch 2: DA heights 110-119
35+
- Epoch 3: DA heights 120-129
36+
37+
### Epoch Boundary Fetching
38+
39+
The `ForcedInclusionRetriever` only returns transactions when queried at the **epoch end** (the last DA height in an epoch):
40+
41+
```go
42+
// When NOT at epoch end -> returns empty transactions
43+
if daHeight != epochEnd {
44+
return &ForcedInclusionEvent{
45+
StartDaHeight: daHeight,
46+
EndDaHeight: daHeight,
47+
Txs: [][]byte{},
48+
}, nil
49+
}
50+
51+
// When AT epoch end -> fetches entire epoch
52+
// Retrieves ALL transactions from epochStart to epochEnd (inclusive)
53+
```
54+
55+
When at an epoch end, the retriever fetches transactions from **all DA blocks in that epoch**:
56+
57+
1. Fetches forced inclusion blobs from `epochStart`
58+
2. Fetches forced inclusion blobs from each height between start and end
59+
3. Fetches forced inclusion blobs from `epochEnd`
60+
4. Returns all transactions as a single `ForcedInclusionEvent`
61+
62+
### Why Epoch-Based?
63+
64+
- **Efficiency**: Reduces the number of DA queries
65+
- **Batching**: Allows processing multiple DA blocks worth of transactions together
66+
- **Determinism**: Clear boundaries for when to fetch from DA
67+
- **Gas optimization**: Fewer DA reads means lower operational costs
68+
69+
## Checkpoint System
70+
71+
### Purpose
72+
73+
The checkpoint system tracks the exact position in the transaction stream to enable crash recovery and ensure no transactions are lost or duplicated.
74+
75+
### Checkpoint Structure
76+
77+
```go
78+
type Checkpoint struct {
79+
// DAHeight is the DA block height currently being processed
80+
DAHeight uint64
81+
82+
// TxIndex is the index of the next transaction to process
83+
// within the DA block's forced inclusion batch
84+
TxIndex uint64
85+
}
86+
```
87+
88+
### How Checkpoints Work
89+
90+
#### 1. Initial State
91+
92+
```
93+
Checkpoint: (DAHeight: 100, TxIndex: 0)
94+
- Ready to fetch epoch starting at DA height 100
95+
```
96+
97+
#### 2. Fetching Transactions
98+
99+
When `GetNextBatch()` is called and we're at an epoch end:
100+
101+
```
102+
Request: GetNextBatch(maxBytes: 1MB)
103+
Action: Fetch all transactions from epoch (DA heights 100-109)
104+
Result: currentBatchTxs = [tx1, tx2, tx3, ..., txN] (from entire epoch)
105+
```
106+
107+
#### 3. Processing Transactions
108+
109+
Transactions are processed incrementally, respecting `maxBytes`:
110+
111+
```
112+
Batch 1: [tx1, tx2] (fits in maxBytes)
113+
Checkpoint: (DAHeight: 100, TxIndex: 2)
114+
115+
Batch 2: [tx3, tx4, tx5]
116+
Checkpoint: (DAHeight: 100, TxIndex: 5)
117+
118+
... continue until all transactions from DA height 100 are consumed
119+
120+
Checkpoint: (DAHeight: 101, TxIndex: 0)
121+
- Moved to next DA height within the same epoch
122+
```
123+
124+
#### 4. Checkpoint Persistence
125+
126+
**Critical**: The checkpoint is persisted to disk **after every batch** of transactions is processed:
127+
128+
```go
129+
if txCount > 0 {
130+
s.checkpoint.TxIndex += txCount
131+
132+
// Move to next DA height when current one is exhausted
133+
if s.checkpoint.TxIndex >= uint64(len(s.currentBatchTxs)) {
134+
s.checkpoint.DAHeight++
135+
s.checkpoint.TxIndex = 0
136+
s.currentBatchTxs = nil
137+
s.SetDAHeight(s.checkpoint.DAHeight)
138+
}
139+
140+
// Persist checkpoint to disk
141+
if err := s.checkpointStore.Save(ctx, s.checkpoint); err != nil {
142+
return nil, fmt.Errorf("failed to save checkpoint: %w", err)
143+
}
144+
}
145+
```
146+
147+
### Crash Recovery Behavior
148+
149+
#### Scenario: Crash Mid-Epoch
150+
151+
**Setup**:
152+
153+
- Epoch 1 spans DA heights 100-109
154+
- At DA height 109, fetched all transactions from the epoch
155+
- Processed transactions up to DA height 105, TxIndex 3
156+
- **Crash occurs**
157+
158+
**On Restart**:
159+
160+
1. **Load Checkpoint**: `(DAHeight: 105, TxIndex: 3)`
161+
2. **Lost Cache**: `currentBatchTxs` is empty (in-memory only)
162+
3. **Attempt Fetch**: `RetrieveForcedIncludedTxs(105)`
163+
4. **Result**: Empty (105 is not an epoch end)
164+
5. **Continue**: Increment DA height, keep trying
165+
6. **Eventually**: Reach DA height 109 (epoch end)
166+
7. **Re-fetch**: Retrieve **entire epoch** again (DA heights 100-109)
167+
8. **Resume**: Use checkpoint to skip already-processed transactions
168+
169+
#### Important Implications
170+
171+
**The entire epoch will be re-fetched after a crash**, even with fine-grained checkpoints.
172+
173+
**Why?**
174+
175+
- Transactions are only available at epoch boundaries
176+
- In-memory cache (`currentBatchTxs`) is lost on restart
177+
- Must wait until the next epoch end to fetch transactions again
178+
179+
**What the checkpoint prevents**:
180+
181+
- ✅ Re-execution of already processed transactions
182+
- ✅ Correct resumption within a DA block's transaction list
183+
- ✅ No transaction loss or duplication
184+
185+
**What the checkpoint does NOT prevent**:
186+
187+
- ❌ Re-fetching the entire epoch from DA
188+
- ❌ Re-validation of previously fetched transactions
189+
190+
### Checkpoint Storage
191+
192+
The checkpoint is stored using a key-value datastore:
193+
194+
```go
195+
// Checkpoint key in the datastore
196+
checkpointKey = ds.NewKey("/based/checkpoint")
197+
198+
// Operations
199+
checkpoint, err := checkpointStore.Load(ctx) // Load from disk
200+
err := checkpointStore.Save(ctx, checkpoint) // Save to disk
201+
err := checkpointStore.Delete(ctx) // Delete from disk
202+
```
203+
204+
The checkpoint is serialized using Protocol Buffers (`pb.SequencerDACheckpoint`) for efficient storage and cross-version compatibility.

0 commit comments

Comments
 (0)