From 6ef0161e29d689bd6458d092cfab443aeb7ffaf3 Mon Sep 17 00:00:00 2001 From: Yiming Zang <50607998+yzang2019@users.noreply.github.com> Date: Tue, 15 Sep 2026 13:46:56 +0000 Subject: [PATCH] Make Receipt and SS write fully async (#4159) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Describe your changes and provide context Both the receipt write and the EVM state store (SS) write were doing slow synchronous work on the block commit path. This makes both of them fully async. ### Receipt store `SetReceipts` used to write the receipt bodies, the `eth_getLogs` index and the version marker inline, and the index commit alone was ~74% of the call. It now hands the block to a background writer and returns. Measured at 2,000 receipts per block, the commit path went from **4.0 ms to 11 µs**. The work still costs the same; it just happens on the writer, where it overlaps with execution instead of serializing against it. - `receipt-store.async-write-buffer` (default 100) bounds how many blocks the store may fall behind. A full queue blocks the caller — that is the back-pressure. - Setting it to `<= 0` keeps writes synchronous, which is the escape hatch if strict read-after-write is wanted. - `LatestVersion()` only advances once a write has actually been applied, so it never advertises a receipt that is not yet readable. It is the watermark a reader follows. ### EVM state store `enqueue_ss` looked async but was dominated by a **synchronous changelog WAL write sitting in front of the queue**. That is also why its queue depth always read 0: queue depth only reveals a slow consumer, and here the producer was the slow side. Under giga that changelog is written every block and never read — crash recovery replays giga's own state WAL via `catchUpTo`, and rollback rewinds SS from its snapshots against that same WAL. So giga now opens SS with `DisableInternalWAL` and the commit-path write is gone. The composite (non-giga) path is untouched and keeps its changelog, which it does need: `ss/composite` rollback replays it to reach versions above a snapshot. ### Interface cleanup `SetLatestVersion` / `SetEarliestVersion` are no longer on the `ReceiptStore` interface. No production code called them — the write path carries the markers, and every external caller was test or benchmark scaffolding. cryptosim's redundant `SetLatestVersion` after each block is deleted for the same reason. ### Bug fixed along the way Draining the pebble async writer on close was nested inside the changelog check: ```go if db.streamHandler != nil { close(db.pendingChanges) db.asyncWriteWG.Wait() ... } ``` With the changelog off, that drain would never run, silently dropping queued blocks on every clean shutdown. The drain is now unconditional, behind a `sync.Once` so `Close` stays idempotent. ### Dashboard `receipt_write_queue_depth` now covers the whole receipt write. The old "ReceiptDB Queue Depth" panel tracked only litt's table queue, which is ~7% of the call, which is why it read 0 while `write_receipts` was a large share of the execution loop. ## Testing performed to validate your change - `sei-db/ledger_db/...`, `sei-db/state_db/...`, `sei-db/bootstrap`, `sei-db/config`, `sei-db/db_engine/pebbledb/...`, `giga/evmonly/...`, `evmrpc/...` and `x/evm/keeper` all pass. - The receipt package passes three repeats under `-race`. - `make dblint` reports 0 issues; `go vet ./...` is clean. New tests: - `TestLittIdxSynchronousWriteBuffer` — with the buffer off, a block is queryable the moment `SetReceipts` returns. - `TestLittIdxWriteBufferBoundsLag` — the buffer is the back-pressure point; the store cannot trail further than it allows. - `TestOpenSSKeepsNoChangelogOfItsOwn` — pins the absence of the SS changelog under giga rather than trusting the config. Verified non-vacuous by re-enabling the flag and watching it fail. Tests that previously relied on read-after-write now wait on `LatestVersion` instead. Worth noting for reviewers: that watermark is necessary but not sufficient as a "my write landed" signal — the bodies land just before the version marker commits, and a block written in parts advances the marker on its first part. The `littidx` helper waits on both. (cherry picked from commit c45517d71484e70972fa2ef499454c69cb8edaf7) --- .../dashboards/gigasim-dashboard.json | 376 +++++++----------- evmrpc/setup_test.go | 14 +- evmrpc/simulate_test.go | 3 +- evmrpc/tests/utils.go | 6 +- .../cryptosim/reciept_store_simulator.go | 4 - sei-db/bench/gigasim/block_generator.go | 28 +- sei-db/bench/gigasim/gigasim.go | 11 +- sei-db/bench/gigasim/gigasim_metrics.go | 12 - sei-db/bench/gigasim/receipt.go | 159 +++++--- sei-db/bench/gigasim/receipt_bloom_test.go | 96 +++++ sei-db/bench/gigasim/receipt_test.go | 26 ++ sei-db/bench/gigasim/receipt_writer.go | 33 +- sei-db/bootstrap/recovery_test.go | 6 +- sei-db/config/giga_config.go | 1 + sei-db/config/receipt_config.go | 17 +- sei-db/config/ss_config.go | 8 + sei-db/db_engine/pebbledb/mvcc/db.go | 48 ++- .../receipt/litt_ctx_internal_test.go | 6 + .../ledger_db/receipt/litt_receipt_store.go | 142 ++++++- .../litt_write_failure_internal_test.go | 180 +++++++++ sei-db/ledger_db/receipt/littidx_test.go | 74 ++++ .../receipt/offline_internal_test.go | 1 + .../receipt/receipt_bench_read_test.go | 5 + sei-db/ledger_db/receipt/receipt_store.go | 26 +- .../ledger_db/receipt/receipt_store_test.go | 6 +- sei-db/ledger_db/receipt/test_helpers_test.go | 12 + sei-db/state_db/giga/state_db.go | 12 +- sei-db/state_db/giga/state_db_replay_test.go | 51 +++ 28 files changed, 980 insertions(+), 383 deletions(-) create mode 100644 sei-db/bench/gigasim/receipt_bloom_test.go create mode 100644 sei-db/ledger_db/receipt/litt_write_failure_internal_test.go diff --git a/docker/monitornode/dashboards/gigasim-dashboard.json b/docker/monitornode/dashboards/gigasim-dashboard.json index b3f01c2688..badc93bd3e 100644 --- a/docker/monitornode/dashboards/gigasim-dashboard.json +++ b/docker/monitornode/dashboards/gigasim-dashboard.json @@ -313,8 +313,8 @@ "kind": "DataQuery", "spec": { "editorMode": "code", - "expr": "litt_control_queue_depth{table=\"receipts\"}", - "legendFormat": "control loop", + "expr": "receipt_write_queue_depth", + "legendFormat": "receipt write (whole write)", "range": true }, "version": "v0" @@ -334,24 +334,45 @@ "kind": "DataQuery", "spec": { "editorMode": "code", - "expr": "litt_flush_queue_depth{table=\"receipts\"}", - "legendFormat": "flush loop", + "expr": "litt_control_queue_depth{table=\"receipts\"}", + "legendFormat": "litt control loop", "range": true }, "version": "v0" }, "refId": "B" } + }, + { + "kind": "PanelQuery", + "spec": { + "hidden": false, + "query": { + "datasource": { + "name": "PBFA97CFB590B2093" + }, + "group": "prometheus", + "kind": "DataQuery", + "spec": { + "editorMode": "code", + "expr": "litt_flush_queue_depth{table=\"receipts\"}", + "legendFormat": "litt flush loop", + "range": true + }, + "version": "v0" + }, + "refId": "C" + } } ], "queryOptions": {}, "transformations": [] } }, - "description": "LittDB's write path for receipt bodies. Receipt writes block on the control loop when it fills.", + "description": "Blocks waiting for the receipt writer. SetReceipts queues the whole write — bodies, log index and version marker — and returns, so this depth is the depth of the receipt write itself. The litt series are the table queues downstream of it.", "id": 23, "links": [], - "title": "ReceiptDB Queue Depth", + "title": "Receipt Write Queue Depth", "vizConfig": { "group": "timeseries", "kind": "VizConfig", @@ -9050,91 +9071,6 @@ } } }, - "panel-104": { - "kind": "Panel", - "spec": { - "data": { - "kind": "QueryGroup", - "spec": { - "queries": [ - { - "kind": "PanelQuery", - "spec": { - "hidden": false, - "query": { - "datasource": { - "name": "PBFA97CFB590B2093" - }, - "group": "prometheus", - "kind": "DataQuery", - "spec": { - "editorMode": "code", - "expr": "rate(gigasim_receipt_write_phase_duration_seconds_total[$__rate_interval])", - "legendFormat": "{{phase}}", - "range": true - }, - "version": "v0" - }, - "refId": "A" - } - } - ], - "queryOptions": {}, - "transformations": [] - } - }, - "description": "What the main execution loop's write_receipts phase is made of: encoding the receipts, which the benchmark does itself, against handing them to the receipt store.", - "id": 104, - "links": [], - "title": "└ Write Receipts — encode vs store", - "vizConfig": { - "group": "piechart", - "kind": "VizConfig", - "spec": { - "fieldConfig": { - "defaults": { - "color": { - "fixedColor": "#73BF69", - "mode": "palette-classic" - }, - "custom": { - "hideFrom": { - "legend": false, - "tooltip": false, - "viz": false - } - }, - "unit": "percentunit" - }, - "overrides": [] - }, - "options": { - "legend": { - "displayMode": "list", - "placement": "bottom", - "showLegend": true, - "values": [] - }, - "pieType": "pie", - "reduceOptions": { - "calcs": [ - "lastNotNull" - ], - "fields": "", - "values": false - }, - "sort": "desc", - "tooltip": { - "hideZeros": false, - "mode": "single", - "sort": "none" - } - } - }, - "version": "13.2.1" - } - } - }, "panel-106": { "kind": "Panel", "spec": { @@ -9803,7 +9739,7 @@ } } }, - "panel-124": { + "panel-125": { "kind": "Panel", "spec": { "data": { @@ -9822,7 +9758,7 @@ "kind": "DataQuery", "spec": { "editorMode": "code", - "expr": "rate(gigasim_receipt_write_phase_duration_seconds_total[$__rate_interval])\n/ on() group_left()\nrate(gigasim_blocks_processed_total[$__rate_interval])", + "expr": "rate(giga_state_commit_phase_duration_seconds_total[$__rate_interval])\n/ on() group_left()\nrate(gigasim_blocks_processed_total[$__rate_interval])", "legendFormat": "{{phase}}", "range": true }, @@ -9836,10 +9772,10 @@ "transformations": [] } }, - "description": "What the main execution loop's write_receipts phase is made of: encoding the receipts, which the benchmark does itself, against handing them to the receipt store. Divided by the block rate, so each band is the time one block spends there rather than a share, which is what shows a phase getting slower while its share holds steady.", - "id": 124, + "description": "What the main execution loop's commit window is made of, split across the state WAL, the state commit store and the EVM state store. These three are the second series on the top-level pie. Divided by the block rate, so each band is the time one block spends there rather than a share, which is what shows a phase getting slower while its share holds steady.", + "id": 125, "links": [], - "title": "└ Write Receipts — encode vs store (per block)", + "title": "└ State Commit — by store (per block)", "vizConfig": { "group": "timeseries", "kind": "VizConfig", @@ -9914,7 +9850,7 @@ } } }, - "panel-125": { + "panel-126": { "kind": "Panel", "spec": { "data": { @@ -9933,7 +9869,7 @@ "kind": "DataQuery", "spec": { "editorMode": "code", - "expr": "rate(giga_state_commit_phase_duration_seconds_total[$__rate_interval])\n/ on() group_left()\nrate(gigasim_blocks_processed_total[$__rate_interval])", + "expr": "rate(seidb_main_thread_phase_duration_seconds_total[$__rate_interval])\n/ on() group_left()\nrate(gigasim_blocks_processed_total[$__rate_interval])", "legendFormat": "{{phase}}", "range": true }, @@ -9947,10 +9883,10 @@ "transformations": [] } }, - "description": "What the main execution loop's commit window is made of, split across the state WAL, the state commit store and the EVM state store. These three are the second series on the top-level pie. Divided by the block rate, so each band is the time one block spends there rather than a share, which is what shows a phase getting slower while its share holds steady.", - "id": 125, + "description": "What the state commit store's commit_sc phase is made of. Covers the commits themselves and not the gaps between them, so it sums to its parent. Divided by the block rate, so each band is the time one block spends there rather than a share, which is what shows a phase getting slower while its share holds steady.", + "id": 126, "links": [], - "title": "└ State Commit — by store (per block)", + "title": " └ SC — inside the FlatKV commit (per block)", "vizConfig": { "group": "timeseries", "kind": "VizConfig", @@ -10025,7 +9961,7 @@ } } }, - "panel-126": { + "panel-127": { "kind": "Panel", "spec": { "data": { @@ -10044,7 +9980,7 @@ "kind": "DataQuery", "spec": { "editorMode": "code", - "expr": "rate(seidb_main_thread_phase_duration_seconds_total[$__rate_interval])\n/ on() group_left()\nrate(gigasim_blocks_processed_total[$__rate_interval])", + "expr": "rate(ss_evm_commit_phase_duration_seconds_total[$__rate_interval])\n/ on() group_left()\nrate(gigasim_blocks_processed_total[$__rate_interval])", "legendFormat": "{{phase}}", "range": true }, @@ -10058,10 +9994,10 @@ "transformations": [] } }, - "description": "What the state commit store's commit_sc phase is made of. Covers the commits themselves and not the gaps between them, so it sums to its parent. Divided by the block rate, so each band is the time one block spends there rather than a share, which is what shows a phase getting slower while its share holds steady.", - "id": 126, + "description": "What the state commit's enqueue_ss phase is made of. The name is misleading: almost all of it is applying the changesets, not queueing them. Divided by the block rate, so each band is the time one block spends there rather than a share, which is what shows a phase getting slower while its share holds steady.", + "id": 127, "links": [], - "title": " └ SC — inside the FlatKV commit (per block)", + "title": " └ SS — inside the EVM state store commit (per block)", "vizConfig": { "group": "timeseries", "kind": "VizConfig", @@ -10136,7 +10072,7 @@ } } }, - "panel-127": { + "panel-129": { "kind": "Panel", "spec": { "data": { @@ -10155,7 +10091,7 @@ "kind": "DataQuery", "spec": { "editorMode": "code", - "expr": "rate(ss_evm_commit_phase_duration_seconds_total[$__rate_interval])\n/ on() group_left()\nrate(gigasim_blocks_processed_total[$__rate_interval])", + "expr": "rate(gigasim_block_producing_loop_phase_duration_seconds_total[$__rate_interval])\n/ on() group_left()\nrate(gigasim_blocks_processed_total[$__rate_interval])", "legendFormat": "{{phase}}", "range": true }, @@ -10169,10 +10105,10 @@ "transformations": [] } }, - "description": "What the state commit's enqueue_ss phase is made of. The name is misleading: almost all of it is applying the changesets, not queueing them. Divided by the block rate, so each band is the time one block spends there rather than a share, which is what shows a phase getting slower while its share holds steady.", - "id": 127, + "description": "Every phase the goroutine that builds blocks and writes them to the block store passes through, as a share of its wall clock. Fully accounted, the blocked hand-off to the execution loop included, so these total 100%. A large wait_for_execution means execution is the limit, not block production. Divided by the block rate, so each band is the time one block spends there rather than a share, which is what shows a phase getting slower while its share holds steady.", + "id": 129, "links": [], - "title": " └ SS — inside the EVM state store commit (per block)", + "title": "Main Block Producing Loop — Time Spent (per block)", "vizConfig": { "group": "timeseries", "kind": "VizConfig", @@ -10247,7 +10183,7 @@ } } }, - "panel-129": { + "panel-130": { "kind": "Panel", "spec": { "data": { @@ -10266,7 +10202,7 @@ "kind": "DataQuery", "spec": { "editorMode": "code", - "expr": "rate(gigasim_block_producing_loop_phase_duration_seconds_total[$__rate_interval])\n/ on() group_left()\nrate(gigasim_blocks_processed_total[$__rate_interval])", + "expr": "rate(gigasim_blockstore_write_phase_duration_seconds_total[$__rate_interval])\n/ on() group_left()\nrate(gigasim_blocks_processed_total[$__rate_interval])", "legendFormat": "{{phase}}", "range": true }, @@ -10280,10 +10216,10 @@ "transformations": [] } }, - "description": "Every phase the goroutine that builds blocks and writes them to the block store passes through, as a share of its wall clock. Fully accounted, the blocked hand-off to the execution loop included, so these total 100%. A large wait_for_execution means execution is the limit, not block production. Divided by the block rate, so each band is the time one block spends there rather than a share, which is what shows a phase getting slower while its share holds steady.", - "id": 129, + "description": "What the block producing loop's write_block phase is made of: the block, the QC and the AppQC covering it, and the periodic flush behind them. Divided by the block rate, so each band is the time one block spends there rather than a share, which is what shows a phase getting slower while its share holds steady.", + "id": 130, "links": [], - "title": "Main Block Producing Loop — Time Spent (per block)", + "title": "└ BlockStore Write — by record (per block)", "vizConfig": { "group": "timeseries", "kind": "VizConfig", @@ -10358,7 +10294,7 @@ } } }, - "panel-130": { + "panel-131": { "kind": "Panel", "spec": { "data": { @@ -10377,7 +10313,7 @@ "kind": "DataQuery", "spec": { "editorMode": "code", - "expr": "rate(gigasim_blockstore_write_phase_duration_seconds_total[$__rate_interval])\n/ on() group_left()\nrate(gigasim_blocks_processed_total[$__rate_interval])", + "expr": "rate(receipt_store_write_phase_duration_seconds_total[$__rate_interval])", "legendFormat": "{{phase}}", "range": true }, @@ -10391,77 +10327,51 @@ "transformations": [] } }, - "description": "What the block producing loop's write_block phase is made of: the block, the QC and the AppQC covering it, and the periodic flush behind them. Divided by the block rate, so each band is the time one block spends there rather than a share, which is what shows a phase getting slower while its share holds steady.", - "id": 130, + "description": "What the execution loop's write_receipts phase is made of. The receipt bodies go to LittDB asynchronously (litt_put), but the same call also maintains the pebble index that serves eth_getLogs, and that is committed inline: stage_tag_keys builds one key per log address plus one per topic, and commit_index writes them. The index scales with logs and topics rather than with receipts, so it dominates the write.", + "id": 131, "links": [], - "title": "└ BlockStore Write — by record (per block)", + "title": " └ Receipt Store Write — litt vs log index", "vizConfig": { - "group": "timeseries", + "group": "piechart", "kind": "VizConfig", "spec": { "fieldConfig": { "defaults": { "color": { + "fixedColor": "#73BF69", "mode": "palette-classic" }, "custom": { - "axisBorderShow": false, - "axisCenteredZero": false, - "axisColorMode": "text", - "axisLabel": "", - "axisPlacement": "auto", - "barAlignment": 0, - "barWidthFactor": 0.6, - "drawStyle": "line", - "fillOpacity": 100, - "gradientMode": "none", "hideFrom": { "legend": false, "tooltip": false, "viz": false - }, - "insertNulls": false, - "lineInterpolation": "linear", - "lineWidth": 1, - "pointSize": 5, - "scaleDistribution": { - "type": "linear" - }, - "showPoints": "auto", - "showValues": false, - "spanNulls": false, - "stacking": { - "group": "A", - "mode": "normal" - }, - "thresholdsStyle": { - "mode": "off" } }, - "thresholds": { - "mode": "absolute", - "steps": [ - { - "color": "green", - "value": 0 - } - ] - }, - "unit": "s" + "unit": "percentunit" }, "overrides": [] }, "options": { "legend": { - "calcs": [], "displayMode": "list", "placement": "bottom", - "showLegend": true + "showLegend": true, + "values": [] + }, + "pieType": "pie", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false }, + "sort": "desc", "tooltip": { "hideZeros": false, - "mode": "multi", - "sort": "desc" + "mode": "single", + "sort": "none" } } }, @@ -10469,7 +10379,7 @@ } } }, - "panel-131": { + "panel-132": { "kind": "Panel", "spec": { "data": { @@ -10488,7 +10398,7 @@ "kind": "DataQuery", "spec": { "editorMode": "code", - "expr": "rate(receipt_store_write_phase_duration_seconds_total[$__rate_interval])", + "expr": "rate(receipt_store_write_phase_duration_seconds_total[$__rate_interval])\n/ on() group_left()\nrate(gigasim_blocks_processed_total[$__rate_interval])", "legendFormat": "{{phase}}", "range": true }, @@ -10502,51 +10412,77 @@ "transformations": [] } }, - "description": "What the execution loop's write_receipts phase is made of. The receipt bodies go to LittDB asynchronously (litt_put), but the same call also maintains the pebble index that serves eth_getLogs, and that is committed inline: stage_tag_keys builds one key per log address plus one per topic, and commit_index writes them. The index scales with logs and topics rather than with receipts, so it dominates the write.", - "id": 131, + "description": "What the execution loop's write_receipts phase is made of. The receipt bodies go to LittDB asynchronously (litt_put), but the same call also maintains the pebble index that serves eth_getLogs, and that is committed inline: stage_tag_keys builds one key per log address plus one per topic, and commit_index writes them. The index scales with logs and topics rather than with receipts, so it dominates the write. Divided by the block rate, so each band is the time one block spends there rather than a share, which is what shows a phase getting slower while its share holds steady.", + "id": 132, "links": [], - "title": " └ Receipt Store Write — litt vs log index", + "title": " └ Receipt Store Write — litt vs log index (per block)", "vizConfig": { - "group": "piechart", + "group": "timeseries", "kind": "VizConfig", "spec": { "fieldConfig": { "defaults": { "color": { - "fixedColor": "#73BF69", "mode": "palette-classic" }, "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "barWidthFactor": 0.6, + "drawStyle": "line", + "fillOpacity": 100, + "gradientMode": "none", "hideFrom": { "legend": false, "tooltip": false, "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "showValues": false, + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "normal" + }, + "thresholdsStyle": { + "mode": "off" } }, - "unit": "percentunit" + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": 0 + } + ] + }, + "unit": "s" }, "overrides": [] }, "options": { "legend": { + "calcs": [], "displayMode": "list", "placement": "bottom", - "showLegend": true, - "values": [] - }, - "pieType": "pie", - "reduceOptions": { - "calcs": [ - "lastNotNull" - ], - "fields": "", - "values": false + "showLegend": true }, - "sort": "desc", "tooltip": { "hideZeros": false, - "mode": "single", - "sort": "none" + "mode": "multi", + "sort": "desc" } } }, @@ -10554,7 +10490,7 @@ } } }, - "panel-132": { + "panel-133": { "kind": "Panel", "spec": { "data": { @@ -10573,8 +10509,8 @@ "kind": "DataQuery", "spec": { "editorMode": "code", - "expr": "rate(receipt_store_write_phase_duration_seconds_total[$__rate_interval])\n/ on() group_left()\nrate(gigasim_blocks_processed_total[$__rate_interval])", - "legendFormat": "{{phase}}", + "expr": "flatkv_snapshot_write_latency_seconds_count", + "legendFormat": "{{success}}", "range": true }, "version": "v0" @@ -10587,10 +10523,10 @@ "transformations": [] } }, - "description": "What the execution loop's write_receipts phase is made of. The receipt bodies go to LittDB asynchronously (litt_put), but the same call also maintains the pebble index that serves eth_getLogs, and that is committed inline: stage_tag_keys builds one key per log address plus one per topic, and commit_index writes them. The index scales with logs and topics rather than with receipts, so it dominates the write. Divided by the block rate, so each band is the time one block spends there rather than a share, which is what shows a phase getting slower while its share holds steady.", - "id": 132, + "description": "Snapshots FlatKV has written since start, counted by the write that produced each one. A series under success=false is a checkpoint that failed: the writer reports it and carries on, so nothing else marks it.", + "id": 133, "links": [], - "title": " └ Receipt Store Write — litt vs log index (per block)", + "title": "Checkpoints Created", "vizConfig": { "group": "timeseries", "kind": "VizConfig", @@ -10609,7 +10545,7 @@ "barAlignment": 0, "barWidthFactor": 0.6, "drawStyle": "line", - "fillOpacity": 100, + "fillOpacity": 15, "gradientMode": "none", "hideFrom": { "legend": false, @@ -10628,7 +10564,7 @@ "spanNulls": false, "stacking": { "group": "A", - "mode": "normal" + "mode": "none" }, "thresholdsStyle": { "mode": "off" @@ -10643,7 +10579,7 @@ } ] }, - "unit": "s" + "unit": "short" }, "overrides": [] }, @@ -10665,7 +10601,7 @@ } } }, - "panel-133": { + "panel-134": { "kind": "Panel", "spec": { "data": { @@ -10684,8 +10620,8 @@ "kind": "DataQuery", "spec": { "editorMode": "code", - "expr": "flatkv_snapshot_write_latency_seconds_count", - "legendFormat": "{{success}}", + "expr": "sum(rate(pebble_pending_changes_queue_blocked_seconds_total{db=~\".*state_store.*\"}[$__rate_interval])) / sum(rate(gigasim_blocks_processed_total[$__rate_interval]))", + "legendFormat": "blocked per block", "range": true }, "version": "v0" @@ -10698,10 +10634,10 @@ "transformations": [] } }, - "description": "Snapshots FlatKV has written since start, counted by the write that produced each one. A series under success=false is a checkpoint that failed: the writer reports it and carries on, so nothing else marks it.", - "id": 133, + "description": "Seconds the commit path spent waiting for room on the EVM state store's apply queue, per block. Depth is sampled and can miss a queue that fills and drains between samples; this counter integrates every wait. When it accounts for most of enqueue_ss, the store is applying slower than blocks arrive, so the commit path is waiting on the queue rather than on work of its own — and no further pipelining helps, because the bottleneck is downstream of it.", + "id": 134, "links": [], - "title": "Checkpoints Created", + "title": "SS Commit Queue — Blocked Time per Block", "vizConfig": { "group": "timeseries", "kind": "VizConfig", @@ -10745,6 +10681,7 @@ "mode": "off" } }, + "min": 0, "thresholds": { "mode": "absolute", "steps": [ @@ -10754,7 +10691,7 @@ } ] }, - "unit": "short" + "unit": "s" }, "overrides": [] }, @@ -11024,32 +10961,6 @@ "y": 54 } }, - { - "kind": "GridLayoutItem", - "spec": { - "element": { - "kind": "ElementReference", - "name": "panel-104" - }, - "height": 9, - "width": 8, - "x": 0, - "y": 63 - } - }, - { - "kind": "GridLayoutItem", - "spec": { - "element": { - "kind": "ElementReference", - "name": "panel-124" - }, - "height": 9, - "width": 16, - "x": 8, - "y": 63 - } - }, { "kind": "GridLayoutItem", "spec": { @@ -11167,6 +11078,19 @@ "x": 16, "y": 8 } + }, + { + "kind": "GridLayoutItem", + "spec": { + "element": { + "kind": "ElementReference", + "name": "panel-134" + }, + "height": 8, + "width": 8, + "x": 0, + "y": 16 + } } ] } diff --git a/evmrpc/setup_test.go b/evmrpc/setup_test.go index 28dc5a279c..bc7a4261d6 100644 --- a/evmrpc/setup_test.go +++ b/evmrpc/setup_test.go @@ -31,6 +31,7 @@ import ( "github.com/sei-protocol/sei-chain/sei-cosmos/crypto/hd" sdk "github.com/sei-protocol/sei-chain/sei-cosmos/types" sdkerrors "github.com/sei-protocol/sei-chain/sei-cosmos/types/errors" + "github.com/sei-protocol/sei-chain/sei-db/ledger_db/receipt" abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types" "github.com/sei-protocol/sei-chain/sei-tendermint/libs/bytes" tmutils "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" @@ -62,6 +63,12 @@ const MockHeight103 = 103 const MockHeight101 = 101 const MockHeight100 = 100 +// pinReceiptVersions widens a store's queryable window to [1, latest]. These tests seed receipts +// by other means, so nothing has advanced the markers a read is gated on. +func pinReceiptVersions(store receipt.ReceiptStore, latest int64) error { + return receipt.PinVersions(store, 1, latest) +} + // LatestCtxUpgradeName makes the test ctx look like a real chain that has // applied a post-v5.8.0 upgrade. The default Ctx has empty // ClosestUpgradeName and semver.Compare("", "v5.8.0") returns -1 (treated @@ -660,11 +667,9 @@ func init() { } testApp.Commit(context.Background()) if store := EVMKeeper.ReceiptStore(); store != nil { - latest := int64(math.MaxInt64) - if err := store.SetLatestVersion(latest); err != nil { + if err := pinReceiptVersions(store, math.MaxInt64); err != nil { panic(err) } - _ = store.SetEarliestVersion(1) } ctxProvider := func(height int64) sdk.Context { if height == MockHeight2 { @@ -1263,10 +1268,9 @@ func setupLogs() { EVMKeeper.SetEvmOnlyBlockBloom(Ctx, []ethtypes.Bloom{bloom4, bloomTx1}) if store := EVMKeeper.ReceiptStore(); store != nil { - if err := store.SetLatestVersion(MockHeight103); err != nil { + if err := pinReceiptVersions(store, MockHeight103); err != nil { panic(err) } - _ = store.SetEarliestVersion(1) } } diff --git a/evmrpc/simulate_test.go b/evmrpc/simulate_test.go index a89bffbce3..c42e722b76 100644 --- a/evmrpc/simulate_test.go +++ b/evmrpc/simulate_test.go @@ -47,8 +47,7 @@ import ( func primeReceiptStore(t *testing.T, store receipt.ReceiptStore, latest int64) { t.Helper() - require.NoError(t, store.SetLatestVersion(latest)) - require.NoError(t, store.SetEarliestVersion(1)) + require.NoError(t, pinReceiptVersions(store, latest)) } // bcAlwaysFailClient fails every Block call (header resolution uses a single block fetch). diff --git a/evmrpc/tests/utils.go b/evmrpc/tests/utils.go index a367663ea5..54592ac5d9 100644 --- a/evmrpc/tests/utils.go +++ b/evmrpc/tests/utils.go @@ -20,6 +20,7 @@ import ( evmrpcconfig "github.com/sei-protocol/sei-chain/evmrpc/config" "github.com/sei-protocol/sei-chain/sei-cosmos/client" sdk "github.com/sei-protocol/sei-chain/sei-cosmos/types" + "github.com/sei-protocol/sei-chain/sei-db/ledger_db/receipt" abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types" tmproto "github.com/sei-protocol/sei-chain/sei-tendermint/proto/tendermint/types" testkeeper "github.com/sei-protocol/sei-chain/testutil/keeper" @@ -187,11 +188,10 @@ func setupTestServer( } pinStateStoreLatestVersion(a, ctxProvider) if store := a.EvmKeeper.ReceiptStore(); store != nil { - latest := int64(math.MaxInt64) - if err := store.SetLatestVersion(latest); err != nil { + // These tests seed receipts by other means and would otherwise read against an unset window. + if err := receipt.PinVersions(store, 1, math.MaxInt64); err != nil { panic(err) } - _ = store.SetEarliestVersion(1) } return TestServer{EVMServer: s, port: port, mockClient: mockClient, app: a, ctxProvider: ctxProvider} } diff --git a/sei-db/bench/cryptosim/reciept_store_simulator.go b/sei-db/bench/cryptosim/reciept_store_simulator.go index 3a0c5aa117..b1bfdc8d7e 100644 --- a/sei-db/bench/cryptosim/reciept_store_simulator.go +++ b/sei-db/bench/cryptosim/reciept_store_simulator.go @@ -251,10 +251,6 @@ func (r *RecieptStoreSimulator) processBlock(blk *block) { for _, entry := range ringEntries { r.txRing.Push(entry.txHash, blockNumber, entry.contractAddress) } - - if err := r.store.SetLatestVersion(int64(blockNumber)); err != nil { //nolint:gosec - fmt.Printf("failed to update latest version for block %d: %v\n", blockNumber, err) - } } // startReceiptReaders launches dedicated goroutines for receipt-by-hash lookups. diff --git a/sei-db/bench/gigasim/block_generator.go b/sei-db/bench/gigasim/block_generator.go index a073b32ac3..597a151b85 100644 --- a/sei-db/bench/gigasim/block_generator.go +++ b/sei-db/bench/gigasim/block_generator.go @@ -3,13 +3,11 @@ package gigasim import ( "context" "fmt" - "hash" - "golang.org/x/crypto/sha3" "golang.org/x/time/rate" "github.com/sei-protocol/sei-chain/sei-db/common/metrics" - evmtypes "github.com/sei-protocol/sei-chain/x/evm/types" + "github.com/sei-protocol/sei-chain/sei-db/ledger_db/receipt" ) // simulatedBlock is one block's worth of work: the transactions the execution phase runs, the payload @@ -22,8 +20,13 @@ type simulatedBlock struct { // executor pool, so a block's transaction count is also its degree of parallelism. transactions []*transaction - // The receipts written to the receipt store, empty when receipts are disabled. - receipts []*evmtypes.Receipt + // The receipts written to the receipt store, in the form it takes them, empty when receipts are + // disabled. They are marshaled here because execution does not change them and its loop paces + // the run. + receiptRecords []receipt.ReceiptRecord + + // What those records marshaled to, which the run reports as bytes written. + receiptBytes int64 // The transaction bytes the block store persists. These stand in for encoded transactions, which // the block store holds as opaque bytes. @@ -77,7 +80,7 @@ type blockGenerator struct { // The keccak hasher every receipt's bloom is built with, held here because only this goroutine // builds receipts. - bloomHasher hash.Hash + receiptCache *receiptCache // This goroutine's share of a block's critical path: building it and storing it. lifecycle *metrics.PhaseTimer @@ -113,7 +116,7 @@ func newBlockGenerator( blocks: blocks, rateLimiter: rateLimiter, blocksChan: make(chan *simulatedBlock, config.MaxPendingExecutionQueueSize), - bloomHasher: sha3.NewLegacyKeccak256(), + receiptCache: newReceiptCache(), lifecycle: gigasimMetrics.NewBlockProducingTimer(), blockStoreWrite: blockStoreWrite, metrics: gigasimMetrics, @@ -201,8 +204,8 @@ func (g *blockGenerator) buildBlock() (*simulatedBlock, error) { } var receipts *receiptBuffer if g.config.EnableReceiptStore { - receipts = newReceiptBuffer(count, g.bloomHasher) - block.receipts = receipts.receipts + receipts = newReceiptBuffer(count, g.receiptCache) + block.receiptRecords = receipts.records } for i := range count { @@ -214,9 +217,14 @@ func (g *blockGenerator) buildBlock() (*simulatedBlock, error) { block.payload[i] = g.accounts.Rand().Bytes(g.config.BytesPerTransaction) if receipts != nil { - receipts.build(i, g.accounts.Rand(), txn, number) + if err := receipts.build(i, g.accounts.Rand(), txn, number); err != nil { + return nil, err + } } } + if receipts != nil { + block.receiptBytes = receipts.encodedBytes + } // Accounts minted for this block become legal read targets once it is complete. g.accounts.ReportEndOfBlock() diff --git a/sei-db/bench/gigasim/gigasim.go b/sei-db/bench/gigasim/gigasim.go index 0cc2796c70..c931977ba2 100644 --- a/sei-db/bench/gigasim/gigasim.go +++ b/sei-db/bench/gigasim/gigasim.go @@ -18,7 +18,7 @@ import ( crand "github.com/sei-protocol/sei-chain/sei-db/common/rand" "github.com/sei-protocol/sei-chain/sei-db/common/utils" dbconfig "github.com/sei-protocol/sei-chain/sei-db/config" - evmtypes "github.com/sei-protocol/sei-chain/x/evm/types" + "github.com/sei-protocol/sei-chain/sei-db/ledger_db/receipt" ) // GigaSim runs the benchmark, driving generated blocks through the block store, the state DB and the @@ -401,7 +401,7 @@ func (g *GigaSim) finalizeSetupBlock() error { if err := g.blocks.writeBlock(number, payload); err != nil { return err } - if err := g.persistExecutionResults(number, nil, g.accounts.Counters()); err != nil { + if err := g.persistExecutionResults(number, nil, 0, g.accounts.Counters()); err != nil { return err } g.accounts.ReportEndOfBlock() @@ -472,7 +472,7 @@ func (g *GigaSim) halt() { func (g *GigaSim) executeAndRecord(block *simulatedBlock) error { g.executeBlock(block) - if err := g.persistExecutionResults(block.number, block.receipts, block.counters); err != nil { + if err := g.persistExecutionResults(block.number, block.receiptRecords, block.receiptBytes, block.counters); err != nil { return err } @@ -514,12 +514,13 @@ func (g *GigaSim) executeBlock(block *simulatedBlock) { // the reverse leaves committed state whose receipts were dropped. func (g *GigaSim) persistExecutionResults( number int64, - receipts []*evmtypes.Receipt, + records []receipt.ReceiptRecord, + receiptBytes int64, counters identifierCounters, ) error { if g.receipts != nil { g.lifecycle.SetPhase("write_receipts") - if err := g.receipts.writeBlock(number, receipts); err != nil { + if err := g.receipts.writeBlock(number, records, receiptBytes); err != nil { return err } } diff --git a/sei-db/bench/gigasim/gigasim_metrics.go b/sei-db/bench/gigasim/gigasim_metrics.go index c35985c29a..20c6675a17 100644 --- a/sei-db/bench/gigasim/gigasim_metrics.go +++ b/sei-db/bench/gigasim/gigasim_metrics.go @@ -52,7 +52,6 @@ type GigasimMetrics struct { executionLoopPhases *metrics.PhaseTimerFactory blockProducingPhases *metrics.PhaseTimerFactory blockStoreWritePhases *metrics.PhaseTimerFactory - receiptWritePhases *metrics.PhaseTimerFactory pendingExecutionQueue *metrics.QueueMeter } @@ -160,7 +159,6 @@ func NewGigasimMetrics() *GigasimMetrics { executionLoopPhases: metrics.NewPhaseTimerFactory(meter, "gigasim_execution_loop").RecordLatencies(), blockProducingPhases: metrics.NewPhaseTimerFactory(meter, "gigasim_block_producing_loop").RecordLatencies(), blockStoreWritePhases: metrics.NewPhaseTimerFactory(meter, "gigasim_blockstore_write"), - receiptWritePhases: metrics.NewPhaseTimerFactory(meter, "gigasim_receipt_write"), pendingExecutionQueue: metrics.NewQueueMeter(meter, "gigasim_pending_execution"), } } @@ -197,16 +195,6 @@ func (m *GigasimMetrics) NewBlockStoreWriteTimer() *metrics.PhaseTimer { return m.blockStoreWritePhases.Build() } -// NewReceiptWriteTimer returns the timer breaking a receipt write into encoding the receipts and -// handing them to the store. It subdivides the execution loop's write_receipts phase rather than -// adding to it. -func (m *GigasimMetrics) NewReceiptWriteTimer() *metrics.PhaseTimer { - if m == nil || m.receiptWritePhases == nil { - return nil - } - return m.receiptWritePhases.Build() -} - // NewTransactionPhaseTimer returns a phase timer for one executor. Each executor needs its own: a // timer tracks a single thread's current phase. func (m *GigasimMetrics) NewTransactionPhaseTimer() *metrics.PhaseTimer { diff --git a/sei-db/bench/gigasim/receipt.go b/sei-db/bench/gigasim/receipt.go index 1dd6e75d3c..b3824558af 100644 --- a/sei-db/bench/gigasim/receipt.go +++ b/sei-db/bench/gigasim/receipt.go @@ -3,11 +3,13 @@ package gigasim import ( "encoding/binary" "encoding/hex" - "hash" + "fmt" + "github.com/ethereum/go-ethereum/common" ethtypes "github.com/ethereum/go-ethereum/core/types" "github.com/sei-protocol/sei-chain/sei-db/common/keys" crand "github.com/sei-protocol/sei-chain/sei-db/common/rand" + "github.com/sei-protocol/sei-chain/sei-db/ledger_db/receipt" evmtypes "github.com/sei-protocol/sei-chain/x/evm/types" ) @@ -64,10 +66,80 @@ func writeSyntheticTxHash(dst []byte, rand *crand.CannedRandom, blockNumber int6 // topicsPerTransferLog is the Transfer event signature plus its two indexed address topics. const topicsPerTransferLog = 3 +// bloomBits are the three bit positions a value contributes to a log bloom. +type bloomBits [3]uint + +// bloomBitsFor derives the three bits a value sets in a log bloom by mixing its bytes rather than +// hashing them, which no filter could match against. Nothing here reads a log back, so what is kept +// is what reaches the store: the same bits per value, the same spread, and the same bits on a +// rerun of the seed. +func bloomBitsFor(value []byte) bloomBits { + // FNV-1a, for a spread across the bloom's positions that costs a multiply per byte. + const ( + fnvOffset uint64 = 14695981039346656037 + fnvPrime uint64 = 1099511628211 + ) + mixed := fnvOffset + for _, b := range value { + mixed ^= uint64(b) + mixed *= fnvPrime + } + var bits bloomBits + for i := range bits { + bits[i] = uint(mixed & 2047) + mixed >>= 11 + } + return bits +} + +// receiptCache holds what a receipt repeats rather than derives anew: the event signature's bloom +// bits, and the values that follow from a contract address. It is not safe for concurrent use. +type receiptCache struct { + signature bloomBits + contracts map[[keys.AddressLen]byte]contractFields +} + +// contractFields are the per-contract values a receipt repeats and none of its transactions change. +type contractFields struct { + bits bloomBits + hex string +} + +// newReceiptCache returns a cache with the constant inputs already resolved. +func newReceiptCache() *receiptCache { + return &receiptCache{ + signature: bloomBitsFor(erc20TransferEventSignatureBytes[:]), + contracts: make(map[[keys.AddressLen]byte]contractFields), + } +} + +// contract returns an ERC20 contract's bloom bits and hex address, resolving one it has not seen. +func (c *receiptCache) contract(address []byte) contractFields { + var key [keys.AddressLen]byte + copy(key[:], address) + if fields, ok := c.contracts[key]; ok { + return fields + } + fields := contractFields{bits: bloomBitsFor(address), hex: bytesToHex(address)} + c.contracts[key] = fields + return fields +} + +// setBits marks bits in a bloom. +func setBits(bloom *ethtypes.Bloom, bits bloomBits) { + for _, bit := range bits { + bloom[ethtypes.BloomByteLength-1-bit/8] |= byte(1 << (bit % 8)) + } +} + // receiptBuffer holds one block's receipts in a fixed number of allocations: every array a receipt // points into is carved out of a slice the buffer owns. type receiptBuffer struct { - receipts []*evmtypes.Receipt + // The records the store is handed, marshaled as each receipt is built. + records []receipt.ReceiptRecord + + // The total size of what those records marshaled to. + encodedBytes int64 storage []evmtypes.Receipt logs []evmtypes.Log @@ -76,30 +148,29 @@ type receiptBuffer struct { blooms []ethtypes.Bloom data []byte - // The bloom hasher, which belongs to the generator rather than to any one block: it is reset - // before each use, and building one per receipt costs more than the hashing does. - hasher hash.Hash + // What the generator has already resolved about the contract pool, kept across blocks. + cache *receiptCache } // newReceiptBuffer allocates the backing storage for one block of receipts. -func newReceiptBuffer(count int, hasher hash.Hash) *receiptBuffer { +func newReceiptBuffer(count int, cache *receiptCache) *receiptBuffer { return &receiptBuffer{ - receipts: make([]*evmtypes.Receipt, count), - storage: make([]evmtypes.Receipt, count), - logs: make([]evmtypes.Log, count), - logRefs: make([]*evmtypes.Log, count), - topics: make([]string, count*topicsPerTransferLog), - blooms: make([]ethtypes.Bloom, count), - data: make([]byte, count*hashLen), - hasher: hasher, + records: make([]receipt.ReceiptRecord, count), + storage: make([]evmtypes.Receipt, count), + logs: make([]evmtypes.Log, count), + logRefs: make([]*evmtypes.Log, count), + topics: make([]string, count*topicsPerTransferLog), + blooms: make([]ethtypes.Bloom, count), + data: make([]byte, count*hashLen), + cache: cache, } } // build fills in the receipt an ERC20 transfer would leave behind: one Transfer log with two indexed // address topics, and a bloom covering them. The values are synthetic, since the receipt store is // measured on the volume and shape of what it stores rather than on the arithmetic behind it. -func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transaction, blockNumber int64) { - contractAddress := addressFromKey(txn.erc20Contract) +func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transaction, blockNumber int64) error { + contract := b.cache.contract(addressFromKey(txn.erc20Contract)) senderTopic := indexedAddressTopic(addressFromKey(txn.srcAccount)) receiverTopic := indexedAddressTopic(addressFromKey(txn.dstAccount)) @@ -112,11 +183,9 @@ func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transact effectiveGasPrice := receiptGasPriceBase + rand.Int64Range(0, receiptGasPriceSpan) transferAmount := receiptTransferBase + rand.Int64Range(0, receiptTransferSpan) - contractAddressHex := bytesToHex(contractAddress) - bloom := &b.blooms[index] *bloom = ethtypes.Bloom{} - b.addTransferLogToBloom(bloom, contractAddress, senderTopic[:], receiverTopic[:]) + b.addTransferLogToBloom(bloom, contract.bits, senderTopic[:], receiverTopic[:]) topics := b.topics[index*topicsPerTransferLog : (index+1)*topicsPerTransferLog] topics[0] = erc20TransferEventSignatureHex @@ -130,7 +199,7 @@ func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transact log := &b.logs[index] b.logRefs[index] = log *log = evmtypes.Log{ - Address: contractAddressHex, + Address: contract.hex, Topics: topics, Data: amount, Index: 0, @@ -139,13 +208,12 @@ func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transact var txHash [hashLen]byte writeSyntheticTxHash(txHash[:], rand, blockNumber, index) - receipt := &b.storage[index] - b.receipts[index] = receipt + built := &b.storage[index] //nolint:gosec // G115 - benchmark values are bounded well below the conversion limits - *receipt = evmtypes.Receipt{ + *built = evmtypes.Receipt{ TxType: txType, CumulativeGasUsed: uint64(gasUsed + int64(index)*previousGas), - ContractAddress: contractAddressHex, + ContractAddress: contract.hex, TxHashHex: bytesToHex(txHash[:]), GasUsed: uint64(gasUsed), EffectiveGasPrice: uint64(effectiveGasPrice), @@ -153,38 +221,37 @@ func (b *receiptBuffer) build(index int, rand *crand.CannedRandom, txn *transact TransactionIndex: uint32(index), Status: uint32(ethtypes.ReceiptStatusSuccessful), From: bytesToHex(addressFromKey(txn.srcAccount)), - To: contractAddressHex, + To: contract.hex, Logs: b.logRefs[index : index+1], LogsBloom: bloom[:], } + + // Marshaled here rather than on the execution loop, which is what paces the run. The receipt is + // final once built, so nothing downstream changes what this encodes. + encoded, err := built.Marshal() + if err != nil { + return fmt.Errorf("failed to marshal the receipt for transaction %d of block %d: %w", + index, blockNumber, err) + } + b.encodedBytes += int64(len(encoded)) + b.records[index] = receipt.ReceiptRecord{ + TxHash: common.BytesToHash(txHash[:]), + Receipt: built, + ReceiptBytes: encoded, + } + return nil } // addTransferLogToBloom sets the bits a Transfer log contributes: the emitting contract, the event // signature and both indexed topics. func (b *receiptBuffer) addTransferLogToBloom( bloom *ethtypes.Bloom, - contractAddress, senderTopic, receiverTopic []byte, + contractBits bloomBits, senderTopic, receiverTopic []byte, ) { - var digest [hashLen]byte - for _, value := range [4][]byte{ - contractAddress, - erc20TransferEventSignatureBytes[:], - senderTopic, - receiverTopic, - } { - addToBloom(b.hasher, &digest, bloom, value) - } -} - -// addToBloom sets the three bits a value contributes to a bloom filter. -func addToBloom(hasher hash.Hash, digest *[hashLen]byte, bloom *ethtypes.Bloom, value []byte) { - hasher.Reset() - _, _ = hasher.Write(value) - sum := hasher.Sum(digest[:0]) - for i := 0; i < 6; i += 2 { - bit := (uint(sum[i])<<8)&2047 + uint(sum[i+1]) - bloom[ethtypes.BloomByteLength-1-bit/8] |= byte(1 << (bit % 8)) - } + setBits(bloom, contractBits) + setBits(bloom, b.cache.signature) + setBits(bloom, bloomBitsFor(senderTopic)) + setBits(bloom, bloomBitsFor(receiverTopic)) } // addressFromKey takes the address out of an EVM key, which carries it after a one-byte prefix. A diff --git a/sei-db/bench/gigasim/receipt_bloom_test.go b/sei-db/bench/gigasim/receipt_bloom_test.go new file mode 100644 index 0000000000..19d6d5c8e1 --- /dev/null +++ b/sei-db/bench/gigasim/receipt_bloom_test.go @@ -0,0 +1,96 @@ +package gigasim + +import ( + "encoding/binary" + "math/bits" + "testing" + + ethtypes "github.com/ethereum/go-ethereum/core/types" + "github.com/sei-protocol/sei-chain/sei-db/common/keys" + "github.com/stretchr/testify/require" + "golang.org/x/crypto/sha3" +) + +// keccakBloomBits is how a real log bloom picks its bits, which these tests compare against. +func keccakBloomBits(value []byte) bloomBits { + hasher := sha3.NewLegacyKeccak256() + _, _ = hasher.Write(value) + sum := hasher.Sum(nil) + var picked bloomBits + for i := 0; i < 6; i += 2 { + picked[i/2] = (uint(sum[i])<<8)&2047 + uint(sum[i+1]) + } + return picked +} + +// bloomTestValue is a distinct value per seed. The seed is written in rather than folded into every +// byte, which wraps at 256 and would yield far fewer values than asked for. +func bloomTestValue(seed int) []byte { + value := make([]byte, hashLen) + binary.BigEndian.PutUint64(value, uint64(seed)) //nolint:gosec // seeds are small and non-negative + for i := 8; i < len(value); i++ { + value[i] = byte(i * 7) + } + return value +} + +// TestBloomBitsForFillsABloomLikeKeccakDoes pins the property the store is measured on: the bits +// differ from a real bloom's, but a corpus has to set as many of them as keccak would. +func TestBloomBitsForFillsABloomLikeKeccakDoes(t *testing.T) { + const values = 2048 + var mixedSet, keccakSet int + for seed := range values { + value := bloomTestValue(seed) + + var mixed, keccak ethtypes.Bloom + setBits(&mixed, bloomBitsFor(value)) + setBits(&keccak, keccakBloomBits(value)) + + mixedSet += countBloomBits(mixed) + keccakSet += countBloomBits(keccak) + } + // Three bits per value either way, less whatever collides; the collision rates have to agree. + require.InDelta(t, keccakSet, mixedSet, float64(keccakSet)*0.01, + "a mixed bloom must fill to the same density as a keccak one, or the corpus compresses differently") +} + +// TestBloomBitsForIsDeterministic pins that a rerun of the same seed produces the same corpus, which +// is what lets two runs be compared. +func TestBloomBitsForIsDeterministic(t *testing.T) { + for seed := range 64 { + value := bloomTestValue(seed) + require.Equal(t, bloomBitsFor(value), bloomBitsFor(value)) + } +} + +// TestBloomBitsForSeparatesValues pins that blooms do not collapse onto a few bit patterns, which +// would compress better than real ones and flatter the store. +func TestBloomBitsForSeparatesValues(t *testing.T) { + const values = 4096 + seen := make(map[bloomBits]struct{}, values) + for seed := range values { + seen[bloomBitsFor(bloomTestValue(seed))] = struct{}{} + } + require.Greater(t, len(seen), values*99/100, "distinct values must land on distinct bits") +} + +// TestReceiptCacheReturnsWhatItCached pins that a contract resolved once reads back the same. +func TestReceiptCacheReturnsWhatItCached(t *testing.T) { + cache := newReceiptCache() + address := make([]byte, keys.AddressLen) + for i := range address { + address[i] = byte(i) + } + first := cache.contract(address) + require.Equal(t, bytesToHex(address), first.hex) + require.Equal(t, bloomBitsFor(address), first.bits) + require.Equal(t, first, cache.contract(address), "the cached value must match the resolved one") +} + +func countBloomBits(bloom ethtypes.Bloom) int { + total := 0 + for _, b := range bloom { + total += bits.OnesCount8(b) + } + return total +} diff --git a/sei-db/bench/gigasim/receipt_test.go b/sei-db/bench/gigasim/receipt_test.go index 5e89ee5b22..d8c9c11fff 100644 --- a/sei-db/bench/gigasim/receipt_test.go +++ b/sei-db/bench/gigasim/receipt_test.go @@ -3,8 +3,10 @@ package gigasim import ( "testing" + "github.com/ethereum/go-ethereum/common" "github.com/stretchr/testify/require" + "github.com/sei-protocol/sei-chain/sei-db/common/keys" crand "github.com/sei-protocol/sei-chain/sei-db/common/rand" ) @@ -34,6 +36,30 @@ func TestSyntheticTxHashesAreUniqueAcrossPositions(t *testing.T) { } } +// TestBuiltRecordKeysOnItsOwnReceiptHash pins a record's key to the hash inside the receipt it +// carries. The store keys on TxHash, so a record keyed on anything else would hide its receipt. +func TestBuiltRecordKeysOnItsOwnReceiptHash(t *testing.T) { + t.Parallel() + + const count = 8 + buffer := newReceiptBuffer(count, newReceiptCache()) + rand := crand.NewCannedRandom(1<<20, 1337) + txn := &transaction{ + erc20Contract: make([]byte, 1+keys.AddressLen+hashLen), + srcAccount: make([]byte, 1+keys.AddressLen+hashLen), + dstAccount: make([]byte, 1+keys.AddressLen+hashLen), + } + + for index := range count { + require.NoError(t, buffer.build(index, rand, txn, 3)) + + record := buffer.records[index] + require.Equal(t, common.HexToHash(record.Receipt.TxHashHex), record.TxHash, + "the record's key must be the hash its own receipt reports") + require.NotEmpty(t, record.ReceiptBytes, "a record reaches the store already marshaled") + } +} + // A hash is recomputable from its position alone, which is what lets a run's transaction hashes be // derived rather than stored. func TestSyntheticTxHashDependsOnlyOnItsPosition(t *testing.T) { diff --git a/sei-db/bench/gigasim/receipt_writer.go b/sei-db/bench/gigasim/receipt_writer.go index 6f65709e0d..5ace22f9a9 100644 --- a/sei-db/bench/gigasim/receipt_writer.go +++ b/sei-db/bench/gigasim/receipt_writer.go @@ -3,12 +3,9 @@ package gigasim import ( "fmt" - "github.com/ethereum/go-ethereum/common" sdk "github.com/sei-protocol/sei-chain/sei-cosmos/types" - "github.com/sei-protocol/sei-chain/sei-db/common/metrics" "github.com/sei-protocol/sei-chain/sei-db/ledger_db/receipt" tmproto "github.com/sei-protocol/sei-chain/sei-tendermint/proto/tendermint/types" - evmtypes "github.com/sei-protocol/sei-chain/x/evm/types" ) // receiptWriter persists a block's receipts through the production write path. It exists only when @@ -16,10 +13,6 @@ import ( type receiptWriter struct { store receipt.ReceiptStore - // Splits this writer's work into encoding the receipts and handing them to the store, subdividing - // the execution loop's write_receipts phase. Only that loop writes receipts, so one timer serves it. - phases *metrics.PhaseTimer - metrics *GigasimMetrics } @@ -27,7 +20,6 @@ type receiptWriter struct { func newReceiptWriter(store receipt.ReceiptStore, gigasimMetrics *GigasimMetrics) *receiptWriter { return &receiptWriter{ store: store, - phases: gigasimMetrics.NewReceiptWriteTimer(), metrics: gigasimMetrics, } } @@ -38,30 +30,7 @@ func newReceiptWriter(store receipt.ReceiptStore, gigasimMetrics *GigasimMetrics // skipping the call would leave the receipt head behind the ledger and the state for every setup // block — heights recovery takes the minimum of, so a run interrupted during setup over an existing // directory would roll state back to a height the ledger has passed and then refuse to reopen. -func (w *receiptWriter) writeBlock(number int64, receipts []*evmtypes.Receipt) error { - // Closes the phase in flight, so the gap until the next block's receipts is charged to neither. - defer w.phases.Reset() - - w.phases.SetPhase("encode") - var encodedBytes int64 - records := make([]receipt.ReceiptRecord, 0, len(receipts)) - for _, rcpt := range receipts { - // The store accepts pre-marshaled bytes, and marshaling here keeps the cost of producing them - // attributed to the benchmark rather than to the store. - encoded, err := rcpt.Marshal() - if err != nil { - return fmt.Errorf("failed to marshal the receipt for transaction %d of block %d: %w", - rcpt.TransactionIndex, number, err) - } - encodedBytes += int64(len(encoded)) - records = append(records, receipt.ReceiptRecord{ - TxHash: common.HexToHash(rcpt.TxHashHex), - Receipt: rcpt, - ReceiptBytes: encoded, - }) - } - - w.phases.SetPhase("store_write") +func (w *receiptWriter) writeBlock(number int64, records []receipt.ReceiptRecord, encodedBytes int64) error { if err := w.store.SetReceipts(sdk.NewContext(nil, tmproto.Header{Height: number}, false), records); err != nil { return fmt.Errorf("failed to write the receipts for block %d: %w", number, err) } diff --git a/sei-db/bootstrap/recovery_test.go b/sei-db/bootstrap/recovery_test.go index 4c10cc72e5..293b4b92d8 100644 --- a/sei-db/bootstrap/recovery_test.go +++ b/sei-db/bootstrap/recovery_test.go @@ -235,7 +235,11 @@ func TestRecoverStoresAtAZeroTargetLeavesReceiptsAlone(t *testing.T) { func TestFindTargetRecoveryHeightIsZeroWithoutABlockLedger(t *testing.T) { manager, _ := openManager(t, nil) commitBlocks(t, manager, 3) - require.NoError(t, manager.ReceiptDB().SetLatestVersion(3)) + // The version marker rides SetReceipts, so it is off the store's interface; this test stamps a + // head without bodies on purpose. + pinner, ok := manager.ReceiptDB().(receipt.VersionPinner) + require.True(t, ok) + require.NoError(t, pinner.SetLatestVersion(3)) // findTargetRecoveryHeight reads the state and receipt directories offline, so both stores have // to be closed for it. closeStateDB(t, manager) diff --git a/sei-db/config/giga_config.go b/sei-db/config/giga_config.go index dca0f6d107..b97a6bc3bf 100644 --- a/sei-db/config/giga_config.go +++ b/sei-db/config/giga_config.go @@ -44,6 +44,7 @@ func DefaultGigaStorageConfig(homePath string) (*GigaStorageConfig, error) { ssConfig := DefaultStateStoreConfig() ssConfig.EVMDBDirectory = utils.GetEVMStateStorePath(homePath, ssConfig.Backend) ssConfig.ExternalPruning = true + ssConfig.DisableInternalWAL = true receiptConfig := DefaultReceiptStoreConfig() receiptConfig.Backend = gigaReceiptBackend diff --git a/sei-db/config/receipt_config.go b/sei-db/config/receipt_config.go index f8ae9df29b..b0a7cf1d04 100644 --- a/sei-db/config/receipt_config.go +++ b/sei-db/config/receipt_config.go @@ -28,6 +28,10 @@ const ( // littidx eth_getLogs (see ReceiptStoreConfig.LogFilterParallelism). const DefaultReceiptLogFilterParallelism = 16 +// DefaultReceiptAsyncWriteBuffer is the default queue depth for receipt writes. It is small because +// the depth is also how far an unclean exit sets recovery back. +const DefaultReceiptAsyncWriteBuffer = 10 + // ReceiptStoreConfig defines configuration for the receipt store database. type ReceiptStoreConfig struct { // Enable reports whether the receipt store is opened. A node with it off keeps no receipt @@ -46,10 +50,15 @@ type ReceiptStoreConfig struct { // defaults to pebbledb Backend string `mapstructure:"rs-backend"` - // AsyncWriteBuffer defines the async queue length for commits to be applied to receipt store - // Applies only to the pebbledb backend. + // AsyncWriteBuffer defines the async queue length for commits to be applied to receipt store. + // It bounds how many blocks the store may fall behind the chain before a write blocks. + // + // Raising it costs more than memory. The queue is not on disk, so an unclean exit loses it and + // the store comes back that far behind, dragging recovery of every other store down with it; + // the EVM RPC head also trails by the queue's depth. Size it for the burst the writer absorbs. + // // Set <= 0 for synchronous writes. - // defaults to 100 + // defaults to 10 AsyncWriteBuffer int `mapstructure:"async-write-buffer"` // KeepRecent defines the number of versions to keep in receipt store. @@ -98,7 +107,7 @@ func DefaultReceiptStoreConfig() ReceiptStoreConfig { return ReceiptStoreConfig{ Enable: true, Backend: "pebbledb", - AsyncWriteBuffer: DefaultSSAsyncBuffer, + AsyncWriteBuffer: DefaultReceiptAsyncWriteBuffer, KeepRecent: 0, PruneIntervalSeconds: DefaultSSPruneInterval, LogFilterParallelism: DefaultReceiptLogFilterParallelism, diff --git a/sei-db/config/ss_config.go b/sei-db/config/ss_config.go index 0e371816f8..c7ab6cae64 100644 --- a/sei-db/config/ss_config.go +++ b/sei-db/config/ss_config.go @@ -38,6 +38,14 @@ type StateStoreConfig struct { // defaults to 100 AsyncWriteBuffer int `mapstructure:"async-write-buffer"` + // DisableInternalWAL stops the backend from keeping a changelog WAL of its own, so a commit is + // not held up by a log write. It is for an owner that already logs every block and replays that + // log into this store, as giga's StateDB does with its state WAL. + // + // Like ExternalPruning it is set by the code wiring the store into its owner rather than read + // from app.toml. Rollback through ss/composite replays the changelog, so that path must keep it. + DisableInternalWAL bool `mapstructure:"-"` + // KeepRecent defines the number of versions to keep in state store (shared by Cosmos and EVM). // Setting it to 0 means keep everything. // Default to keep the last 100,000 blocks diff --git a/sei-db/db_engine/pebbledb/mvcc/db.go b/sei-db/db_engine/pebbledb/mvcc/db.go index 7e079dc465..7d0bc44ec5 100644 --- a/sei-db/db_engine/pebbledb/mvcc/db.go +++ b/sei-db/db_engine/pebbledb/mvcc/db.go @@ -103,6 +103,9 @@ type Database struct { // Pending changes to be written to the DB pendingChanges chan VersionedChangesets + // Guards the one close of pendingChanges, so Close stays idempotent. + drainOnce sync.Once + // Reports pendingChanges from the writer's side: how full it was when a write needed room, and how // long writes waited when it had none. pendingChangesQueue *seidbmetrics.QueueMeter @@ -149,7 +152,7 @@ func newPebbleOptions(config config.StateStoreConfig, cache *pebble.Cache) *pebb FormatMajorVersion: pebble.FormatVirtualSSTables, L0CompactionThreshold: 2, L0StopWritesThreshold: 1000, - LBaseMaxBytes: 64 << 20, // 64 MB + LBaseMaxBytes: 64 << 20, // 64 MiB MemTableSize: 64 << 20, MemTableStopWritesThreshold: 4, // Let Pebble run several compactions in parallel so it can keep up with @@ -239,23 +242,27 @@ func OpenDB(dataDir string, config config.StateStoreConfig) (types.StateStore, e _ = db.Close() return nil, errors.New("KeepRecent must be non-negative") } - walKeepRecent := changelogKeepRecent(config) - // Snapshot rollback replays the changelog forward from the oldest retained - // snapshot, so count-based pruning must not cut inside that span. The - // snapshot manager prunes this changelog by snapshot version after every - // retention pass and is what actually holds it down; the count below is the - // ceiling for the states that pass does not cover — external snapshot - // pruning, and the stretch before enough snapshots exist to prune. Raising - // the ceiling is what a rollback window costs on disk: roughly one snapshot - // interval of changelog per retained snapshot. - streamHandler, err := wal.NewChangelogWAL(utils.GetChangelogPath(dataDir), wal.Config{ - KeepRecent: walKeepRecent, - PruneInterval: time.Duration(config.PruneIntervalSeconds) * time.Second, - }) - if err != nil { - return nil, err + // An owner that logs every block replays it into this store, leaving the changelog here written + // and never read. + if !config.DisableInternalWAL { + walKeepRecent := changelogKeepRecent(config) + // Snapshot rollback replays the changelog forward from the oldest retained + // snapshot, so count-based pruning must not cut inside that span. The + // snapshot manager prunes this changelog by snapshot version after every + // retention pass and is what actually holds it down; the count below is the + // ceiling for the states that pass does not cover — external snapshot + // pruning, and the stretch before enough snapshots exist to prune. Raising + // the ceiling is what a rollback window costs on disk: roughly one snapshot + // interval of changelog per retained snapshot. + streamHandler, err := wal.NewChangelogWAL(utils.GetChangelogPath(dataDir), wal.Config{ + KeepRecent: walKeepRecent, + PruneInterval: time.Duration(config.PruneIntervalSeconds) * time.Second, + }) + if err != nil { + return nil, err + } + database.streamHandler = streamHandler } - database.streamHandler = streamHandler database.asyncWriteWG.Add(1) go database.writeAsyncInBackground() @@ -395,12 +402,15 @@ func (db *Database) Close() error { db.metricsCancel() } - if db.streamHandler != nil { + // Owed whether or not a changelog is kept, the queued blocks being only in memory. The channel + // is left in place so a send after close still panics rather than blocking on a nil one. + db.drainOnce.Do(func() { // First, stop accepting new pending changes and drain the worker close(db.pendingChanges) // Wait for the async writes to finish db.asyncWriteWG.Wait() - // Now close the WAL stream + }) + if db.streamHandler != nil { _ = db.streamHandler.Close() db.streamHandler = nil } diff --git a/sei-db/ledger_db/receipt/litt_ctx_internal_test.go b/sei-db/ledger_db/receipt/litt_ctx_internal_test.go index 1f10a1341c..6e4cdbd467 100644 --- a/sei-db/ledger_db/receipt/litt_ctx_internal_test.go +++ b/sei-db/ledger_db/receipt/litt_ctx_internal_test.go @@ -77,6 +77,7 @@ func TestBlockLogsReturnsCanceledContextBeforeScanning(t *testing.T) { topic := common.HexToHash("0xdef1") txHash, rcpt := littCtxTestReceipt(1, 0, addr, topic, 1) require.NoError(t, s.SetReceipts(newTestCtxAtHeight(1), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}})) + requireReceiptVersion(t, s, 1) ctx, cancel := context.WithCancel(context.Background()) cancel() @@ -95,6 +96,7 @@ func TestCandidateBlockLogsReturnsCanceledContextBeforeTx(t *testing.T) { topic := common.HexToHash("0xdef2") txHash, rcpt := littCtxTestReceipt(2, 0, addr, topic, 1) require.NoError(t, s.SetReceipts(newTestCtxAtHeight(2), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}})) + requireReceiptVersion(t, s, 2) candidates, err := s.blockTagCandidates(2, criteriaTagGroups(filters.FilterCriteria{})) require.NoError(t, err) @@ -121,6 +123,7 @@ func TestCandidateBlockLogsCancelsMidLoopOverLogs(t *testing.T) { topic := common.HexToHash("0xdef3") txHash, rcpt := littCtxTestReceipt(3, 0, addr, topic, 5) require.NoError(t, s.SetReceipts(newTestCtxAtHeight(3), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}})) + requireReceiptVersion(t, s, 3) candidates, err := s.blockTagCandidates(3, criteriaTagGroups(filters.FilterCriteria{})) require.NoError(t, err) @@ -147,6 +150,7 @@ func TestCandidateBlockLogsTripsBudgetMidLoopOverLogs(t *testing.T) { topic := common.HexToHash("0xdef4") txHash, rcpt := littCtxTestReceipt(4, 0, addr, topic, 2) require.NoError(t, s.SetReceipts(newTestCtxAtHeight(4), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}})) + requireReceiptVersion(t, s, 4) candidates, err := s.blockTagCandidates(4, criteriaTagGroups(filters.FilterCriteria{})) require.NoError(t, err) @@ -173,6 +177,7 @@ func TestFilterLogsByTagsPreCanceledContextReturnsEmptyFast(t *testing.T) { for block := uint64(1); block <= 5; block++ { txHash, rcpt := littCtxTestReceipt(block, 0, addr, topic, 1) require.NoError(t, s.SetReceipts(newTestCtxAtHeight(block), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}})) + requireReceiptVersion(t, s, int64(block)) } ctx, cancel := context.WithCancel(context.Background()) @@ -196,6 +201,7 @@ func TestFilterLogsThreadsSDKContext(t *testing.T) { topic := common.HexToHash("0xdef6") txHash, rcpt := littCtxTestReceipt(6, 0, addr, topic, 1) require.NoError(t, s.SetReceipts(newTestCtxAtHeight(6), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}})) + requireReceiptVersion(t, s, 6) crit := filters.FilterCriteria{Addresses: []common.Address{addr}} diff --git a/sei-db/ledger_db/receipt/litt_receipt_store.go b/sei-db/ledger_db/receipt/litt_receipt_store.go index a645587369..9155b47e58 100644 --- a/sei-db/ledger_db/receipt/litt_receipt_store.go +++ b/sei-db/ledger_db/receipt/litt_receipt_store.go @@ -62,6 +62,9 @@ import ( // - unset: the background pruner below keeps the last KeepRecent blocks. // - set: the StorageGarbageCollector prunes through the gc.PrunableStore // implementation in litt_receipt_gc.go, and startPruning stands down. +// +// Writes are applied in the background, so a receipt is not necessarily readable when SetReceipts +// returns. LatestVersion is the watermark of what has been applied; Close waits for the queue. type littReceiptStore struct { values litt.DB receipts litt.Table @@ -79,12 +82,33 @@ type littReceiptStore struct { backgroundWg sync.WaitGroup closeOnce sync.Once - // Breaks a write into its stages. The receipt bodies go to litt asynchronously while the log index - // is committed inline, so which of the two a slow write is in is not otherwise visible. Only the - // commit path writes, so one timer serves the store. + // Breaks a write into its stages. Only the writer goroutine records, so one timer serves the store. writePhases *seidbmetrics.PhaseTimer + + // Receipt writes waiting to be applied, and the meter for time spent waiting on a full queue. A + // whole write is queued, so the depth is the receipt write's own. Nil means writes apply inline. + writes chan receiptWrite + + // Orders admitting a write against shutting the writer down, so none is accepted into a queue + // that will not be drained. queueWrite holds it shared; Close takes it exclusively. + admission sync.RWMutex + closing bool + + writeQueue *seidbmetrics.QueueMeter + writeErr atomic.Pointer[error] + stopSampling context.CancelFunc +} + +// receiptWrite is one block's receipts, waiting to be applied. +type receiptWrite struct { + height int64 + receipts []ReceiptRecord } +// writeQueueSampleIntervalSeconds is how often the write queue's depth is read. Sampling on a timer +// rather than at each send keeps the reading unbiased by the send rate. +const writeQueueSampleIntervalSeconds = 1 + var _ ReceiptStore = (*littReceiptStore)(nil) var ( @@ -213,7 +237,19 @@ func newLittReceiptStore(cfg dbconfig.ReceiptStoreConfig, storeKey sdk.StoreKey) return nil, fmt.Errorf("failed to open receipt log index: %w", err) } s.index = index - s.writePhases = seidbmetrics.NewPhaseTimer(otel.Meter("seidb_receipt"), "receipt_store_write") + + receiptMeter := otel.Meter("seidb_receipt") + s.writePhases = seidbmetrics.NewPhaseTimer(receiptMeter, "receipt_store_write") + if cfg.AsyncWriteBuffer > 0 { + s.writes = make(chan receiptWrite, cfg.AsyncWriteBuffer) + s.writeQueue = seidbmetrics.NewQueueMeter(receiptMeter, "receipt_write") + s.startWriter() + + samplingCtx, stopSampling := context.WithCancel(context.Background()) + s.stopSampling = stopSampling + s.writeQueue.SampleDepth(samplingCtx, writeQueueSampleIntervalSeconds, + func() int { return len(s.writes) }) + } s.latestVersion.Store(s.readMeta(receiptLatestVersionKey)) s.earliestVersion.Store(s.readMeta(receiptEarliestVersionKey)) @@ -301,17 +337,47 @@ func (s *littReceiptStore) belowRetentionFloor(blockNumber uint64) bool { return earliest > 0 && blockNumber < uint64(earliest) //nolint:gosec // earliest is non-negative } +// SetReceipts hands the block's receipts to the writer, blocking only when the queue is full, or +// applies them inline when AsyncWriteBuffer is off. Once a queued write has failed it takes no +// further block and returns that failure. func (s *littReceiptStore) SetReceipts(ctx sdk.Context, receipts []ReceiptRecord) error { + if s.writes == nil { + return s.applyReceipts(ctx.BlockHeight(), receipts) + } + if err := s.writeFailure(); err != nil { + return err + } + return s.queueWrite(receiptWrite{height: ctx.BlockHeight(), receipts: receipts}) +} + +// ErrStoreClosed is returned by a write the store can no longer apply, the writer having stopped. +var ErrStoreClosed = errors.New("receipt store is closed") + +// queueWrite hands a write to the writer, waiting for room when the queue is full and refusing once +// the store is closing. +func (s *littReceiptStore) queueWrite(write receiptWrite) error { + // Held across the send, not merely to read the flag: Close takes it exclusively before stopping + // the writer, so a write admitted here always reaches a writer that is still running. + s.admission.RLock() + defer s.admission.RUnlock() + if s.closing { + return ErrStoreClosed + } + seidbmetrics.Send(s.writeQueue, s.writes, write) + return nil +} + +// applyReceipts writes a block's receipt bodies, log index and version marker. The bodies go to +// litt first, so an indexed block always has its values written. +func (s *littReceiptStore) applyReceipts(height int64, receipts []ReceiptRecord) error { blockNumbers, receiptsByBlock := groupReceiptRecordsByBlock(receipts) if len(blockNumbers) == 0 { - return s.SetLatestVersion(ctx.BlockHeight()) + return s.SetLatestVersion(height) } // Closes the stage in flight, so the gap until the next write is charged to neither. defer s.writePhases.Reset() - // Receipt values go to litt first; the index batch (tag keys + version - // meta) commits after, so an indexed block always has its values written. batch := s.index.NewBatch() defer func() { _ = batch.Close() }() @@ -421,6 +487,53 @@ func (s *littReceiptStore) FilterLogs(ctx sdk.Context, fromBlock, toBlock uint64 return s.filterLogsByTags(reqCtx, fromBlock, toBlock, crit, budget) } +// startWriter applies queued receipt writes in the order they were enqueued, until the store closes. +// It drains what it holds before returning, so a clean shutdown applies them all and an unclean exit +// loses the queue. +func (s *littReceiptStore) startWriter() { + s.backgroundWg.Add(1) + go func() { + defer s.backgroundWg.Done() + for { + select { + case write := <-s.writes: + s.applyWrite(write) + case <-s.stopBackground: + for { + select { + case write := <-s.writes: + s.applyWrite(write) + default: + return + } + } + } + } + }() +} + +// applyWrite performs one queued write, keeping the first failure for its callers to collect. +// Nothing is applied after a failure: a later block carries its own version marker and would publish +// a head above one whose receipts were never written. +func (s *littReceiptStore) applyWrite(write receiptWrite) { + if s.writeFailure() != nil { + return + } + if err := s.applyReceipts(write.height, write.receipts); err != nil { + logger.Error("failed to write receipts", "height", write.height, "err", err) + s.writeErr.CompareAndSwap(nil, &err) + } +} + +// writeFailure returns the first failure a queued write hit. It latches, so every later caller sees +// it rather than the first to ask consuming it. +func (s *littReceiptStore) writeFailure() error { + if err := s.writeErr.Load(); err != nil { + return *err + } + return nil +} + // startFlusher bounds litt durability lag to littFlushInterval from a // background goroutine so block commit never waits on an fsync. func (s *littReceiptStore) startFlusher() { @@ -445,10 +558,23 @@ func (s *littReceiptStore) startFlusher() { func (s *littReceiptStore) Close() error { var err error s.closeOnce.Do(func() { + // Exclusive and before the writer stops, so it takes effect only once the writes already + // admitted have been handed over. + s.admission.Lock() + s.closing = true + s.admission.Unlock() + + if s.stopSampling != nil { + s.stopSampling() + } close(s.stopBackground) + // The writer drains what it holds before returning, so this is where queued writes land. s.backgroundWg.Wait() + err = s.writeFailure() // litt's Close flushes, so the last sub-interval of writes is durable. - err = s.values.Close() + if valuesErr := s.values.Close(); err == nil { + err = valuesErr + } if indexErr := s.index.Close(); err == nil { err = indexErr } diff --git a/sei-db/ledger_db/receipt/litt_write_failure_internal_test.go b/sei-db/ledger_db/receipt/litt_write_failure_internal_test.go new file mode 100644 index 0000000000..e53242cd06 --- /dev/null +++ b/sei-db/ledger_db/receipt/litt_write_failure_internal_test.go @@ -0,0 +1,180 @@ +package receipt + +import ( + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + dbtypes "github.com/sei-protocol/sei-chain/sei-db/db_engine/types" + "github.com/stretchr/testify/require" +) + +var errIndexCommit = errors.New("injected index commit failure") + +// failingIndex is the store's log index with its batch commits made to fail on demand. Holding a +// commit open is what lets a test queue a block behind the one that is failing. +type failingIndex struct { + dbtypes.KeyValueDB + failing atomic.Bool + entered chan struct{} // closed once a failing commit has been reached + enteredOnce sync.Once // more than one commit may fail, and entered closes for the first + release chan struct{} // closed to let that commit return its error +} + +func (f *failingIndex) NewBatch() dbtypes.Batch { + return &failingBatch{Batch: f.KeyValueDB.NewBatch(), index: f} +} + +type failingBatch struct { + dbtypes.Batch + index *failingIndex +} + +func (b *failingBatch) Commit(opts dbtypes.WriteOptions) error { + if b.index.failing.Load() { + b.index.enteredOnce.Do(func() { close(b.index.entered) }) + <-b.index.release + return errIndexCommit + } + return b.Batch.Commit(opts) +} + +// TestWriteFailureHoldsTheHeadAgainstAQueuedBlock covers what a failed write owes the blocks queued +// behind it: applying one would publish a head above the block that never landed. The follower is +// queued while the failing commit is held, since SetReceipts refuses blocks once the failure shows. +func TestWriteFailureHoldsTheHeadAgainstAQueuedBlock(t *testing.T) { + s, closeStore := setupLittCtxStore(t) + defer closeStore() + + addr := common.HexToAddress("0xfa11") + topic := common.HexToHash("0xfa12") + + index := &failingIndex{ + KeyValueDB: s.index, + entered: make(chan struct{}), + release: make(chan struct{}), + } + s.index = index + + // Block 1 lands, so there is a real head for the failure to hold. + writeOneReceipt(t, s, 1, addr, topic) + requireReceiptVersion(t, s, 1) + + // Block 2 reaches its commit and stops there, still holding the writer. + index.failing.Store(true) + writeOneReceipt(t, s, 2, addr, topic) + <-index.entered + + // Block 3 would commit cleanly and carry a marker naming it the head. Queued now, while block 2 + // is mid-commit, it is past the refusal in SetReceipts and only the writer can hold it back. + index.failing.Store(false) + writeOneReceipt(t, s, 3, addr, topic) + + close(index.release) + + // Close drains, so the writer has decided about block 3 by the time this returns. + require.ErrorIs(t, s.Close(), errIndexCommit) + require.Equal(t, int64(1), s.LatestVersion(), + "the head must not move past a block whose receipts were never written") +} + +// TestWriteFailureLatches covers the failure reaching every later caller rather than only the first +// to ask. +func TestWriteFailureLatches(t *testing.T) { + s, closeStore := setupLittCtxStore(t) + defer closeStore() + + addr := common.HexToAddress("0xfa21") + topic := common.HexToHash("0xfa22") + + index := &failingIndex{ + KeyValueDB: s.index, + entered: make(chan struct{}), + release: make(chan struct{}), + } + s.index = index + close(index.release) + + index.failing.Store(true) + writeOneReceipt(t, s, 1, addr, topic) + require.Eventually(t, func() bool { return s.writeFailure() != nil }, 5*time.Second, time.Millisecond) + + txHash, rcpt := littCtxTestReceipt(2, 0, addr, topic, 1) + require.ErrorIs(t, s.SetReceipts(newTestCtxAtHeight(2), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}}), + errIndexCommit, "a commit after a failed write must be refused rather than queued") + require.ErrorIs(t, s.writeFailure(), errIndexCommit, "reading the failure must not consume it") + require.ErrorIs(t, s.Close(), errIndexCommit, "Close must report it too") +} + +// TestWriteAfterCloseIsRefused covers a commit arriving after shutdown, which the writer is no +// longer there to apply. +func TestWriteAfterCloseIsRefused(t *testing.T) { + s, _ := setupLittCtxStore(t) + require.NoError(t, s.Close()) + + txHash, rcpt := littCtxTestReceipt(1, 0, common.HexToAddress("0xfa31"), common.HexToHash("0xfa32"), 1) + err := s.SetReceipts(newTestCtxAtHeight(1), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}}) + require.ErrorIs(t, err, ErrStoreClosed) +} + +// TestWriteAfterCloseIsRefusedWithAFullQueue is the same refusal with no room left to send into, +// which would otherwise block forever. +func TestWriteAfterCloseIsRefusedWithAFullQueue(t *testing.T) { + s, _ := setupLittCtxStore(t) + require.NoError(t, s.Close()) + + // Leftovers with no writer behind them: the send has nowhere to go and nobody to take it. + for len(s.writes) < cap(s.writes) { + s.writes <- receiptWrite{height: 1} + } + + txHash, rcpt := littCtxTestReceipt(1, 0, common.HexToAddress("0xfa41"), common.HexToHash("0xfa42"), 1) + done := make(chan error, 1) + go func() { + done <- s.SetReceipts(newTestCtxAtHeight(1), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}}) + }() + select { + case err := <-done: + require.ErrorIs(t, err, ErrStoreClosed) + case <-time.After(5 * time.Second): + t.Fatal("a write into a full queue on a closed store never returned") + } +} + +// TestWriteRacingCloseIsEitherAppliedOrRefused covers a write admitted while Close is running, +// which the two tests above cannot reach. A write reporting success must have been applied. +func TestWriteRacingCloseIsEitherAppliedOrRefused(t *testing.T) { + for attempt := range 50 { + s, _ := setupLittCtxStore(t) + + addr := common.HexToAddress("0xfa51") + topic := common.HexToHash("0xfa52") + txHash, rcpt := littCtxTestReceipt(1, 0, addr, topic, 1) + + started := make(chan struct{}) + result := make(chan error, 1) + go func() { + close(started) + result <- s.SetReceipts(newTestCtxAtHeight(1), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}}) + }() + <-started + require.NoError(t, s.Close()) + + if err := <-result; err != nil { + require.ErrorIs(t, err, ErrStoreClosed, "attempt %d", attempt) + continue + } + // Accepted, so the writer must have applied it before Close let the writer go. + require.Equal(t, int64(1), s.LatestVersion(), + "attempt %d: a write that reported success must have been applied", attempt) + } +} + +func writeOneReceipt(t *testing.T, s *littReceiptStore, block uint64, addr common.Address, topic common.Hash) { + t.Helper() + txHash, rcpt := littCtxTestReceipt(block, 0, addr, topic, 1) + require.NoError(t, s.SetReceipts(newTestCtxAtHeight(block), []ReceiptRecord{{TxHash: txHash, Receipt: rcpt}})) +} diff --git a/sei-db/ledger_db/receipt/littidx_test.go b/sei-db/ledger_db/receipt/littidx_test.go index d4f5fa23ad..3d123159fb 100644 --- a/sei-db/ledger_db/receipt/littidx_test.go +++ b/sei-db/ledger_db/receipt/littidx_test.go @@ -2,9 +2,12 @@ package receipt_test import ( "fmt" + "slices" "testing" + "time" "github.com/ethereum/go-ethereum/common" + ethtypes "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth/filters" storetypes "github.com/sei-protocol/sei-chain/sei-cosmos/store/types" "github.com/sei-protocol/sei-chain/sei-cosmos/testutil" @@ -15,6 +18,62 @@ import ( "github.com/stretchr/testify/require" ) +// TestLittIdxSynchronousWriteBuffer pins the AsyncWriteBuffer <= 0 case: the write is applied on the +// caller, so the block is queryable the moment SetReceipts returns. +func TestLittIdxSynchronousWriteBuffer(t *testing.T) { + storeKey := storetypes.NewKVStoreKey("evm") + tkey := storetypes.NewTransientStoreKey("evm_transient") + ctx := testutil.DefaultContext(storeKey, tkey).WithBlockHeight(1) + cfg := dbconfig.DefaultReceiptStoreConfig() + cfg.Backend = "littidx" + cfg.DBDirectory = t.TempDir() + cfg.KeepRecent = 0 + cfg.AsyncWriteBuffer = 0 + + store, err := receipt.NewReceiptStore(cfg, storeKey) + require.NoError(t, err) + t.Cleanup(func() { _ = store.Close() }) + + addr := common.HexToAddress("0xabc") + record := litReceipt(1, 0, addr, common.HexToHash("0xdead")) + require.NoError(t, store.SetReceipts(ctx, []receipt.ReceiptRecord{record})) + + require.Equal(t, int64(1), store.LatestVersion()) + got, err := store.GetReceipt(ctx, record.TxHash) + require.NoError(t, err) + require.Equal(t, record.Receipt.TxHashHex, got.TxHashHex) +} + +// TestLittIdxWriteBufferBoundsLag pins that the buffer is the back-pressure point: with room for one +// block, a writer cannot get further than the buffer ahead of what has been applied. +func TestLittIdxWriteBufferBoundsLag(t *testing.T) { + storeKey := storetypes.NewKVStoreKey("evm") + tkey := storetypes.NewTransientStoreKey("evm_transient") + ctx := testutil.DefaultContext(storeKey, tkey).WithBlockHeight(1) + cfg := dbconfig.DefaultReceiptStoreConfig() + cfg.Backend = "littidx" + cfg.DBDirectory = t.TempDir() + cfg.KeepRecent = 0 + cfg.AsyncWriteBuffer = 1 + + store, err := receipt.NewReceiptStore(cfg, storeKey) + require.NoError(t, err) + t.Cleanup(func() { _ = store.Close() }) + + addr := common.HexToAddress("0xabc") + const blocks = 8 + for block := uint64(1); block <= blocks; block++ { + record := litReceipt(block, 0, addr, common.HexToHash("0xdead")) + require.NoError(t, store.SetReceipts(ctx.WithBlockHeight(int64(block)), //nolint:gosec // small test heights + []receipt.ReceiptRecord{record})) + // One queued block plus the one in flight is as far as the store may trail. + require.GreaterOrEqual(t, store.LatestVersion(), int64(block)-2) //nolint:gosec // small test heights + } + + require.Eventually(t, func() bool { return store.LatestVersion() == blocks }, + 5*time.Second, time.Millisecond) +} + func setupLittIdx(t *testing.T, dir string) (receipt.ReceiptStore, sdk.Context) { t.Helper() return setupLittIdxPar(t, dir, dbconfig.DefaultReceiptLogFilterParallelism) @@ -64,6 +123,21 @@ func litReceipt(block uint64, txIndex uint32, addr common.Address, topics ...com func writeLitBlock(t *testing.T, store receipt.ReceiptStore, ctx sdk.Context, block uint64, records ...receipt.ReceiptRecord) { t.Helper() require.NoError(t, store.SetReceipts(ctx.WithBlockHeight(int64(block)), records)) //nolint:gosec // small test heights + if len(records) == 0 { + return + } + // A write puts its bodies in litt before it commits its log index, so a readable receipt does not + // mean a queryable one. LatestVersion does not close that gap either: a block written in parts + // does not advance it past the first part. Waiting for the last record's log covers both stages. + last := records[len(records)-1].TxHash + require.Eventually(t, func() bool { + //nolint:gosec // small test heights + logs, err := store.FilterLogs(ctx, block, block, filters.FilterCriteria{}, nil) + if err != nil { + return false + } + return slices.ContainsFunc(logs, func(l *ethtypes.Log) bool { return l.TxHash == last }) + }, 5*time.Second, time.Millisecond) } func TestLittIdxReadWrite(t *testing.T) { diff --git a/sei-db/ledger_db/receipt/offline_internal_test.go b/sei-db/ledger_db/receipt/offline_internal_test.go index ac030667ca..2fd51092e9 100644 --- a/sei-db/ledger_db/receipt/offline_internal_test.go +++ b/sei-db/ledger_db/receipt/offline_internal_test.go @@ -36,6 +36,7 @@ func writeLittIdxReceipts(t *testing.T, dir string, blocks uint64) { []common.Hash{topic})} //nolint:gosec // small test heights require.NoError(t, store.SetReceipts(ctx.WithBlockHeight(int64(block)), []ReceiptRecord{record})) + requireReceiptVersion(t, store, int64(block)) //nolint:gosec // small test heights } require.NoError(t, store.Close()) } diff --git a/sei-db/ledger_db/receipt/receipt_bench_read_test.go b/sei-db/ledger_db/receipt/receipt_bench_read_test.go index 0605370167..4d9e002011 100644 --- a/sei-db/ledger_db/receipt/receipt_bench_read_test.go +++ b/sei-db/ledger_db/receipt/receipt_bench_read_test.go @@ -6,6 +6,7 @@ import ( "sync" "sync/atomic" "testing" + "time" "github.com/ethereum/go-ethereum/common" ethtypes "github.com/ethereum/go-ethereum/core/types" @@ -271,6 +272,10 @@ func setupReadBenchmark(b *testing.B, backend string, blocks, receiptsPerBlock, if err := store.SetReceipts(ctx.WithBlockHeight(int64(blockNumber)), batch); err != nil { b.Fatalf("failed to write block %d: %v", blockNumber, err) } + // Seeding outruns the writer, so wait for the block to be published before the next one. + for store.LatestVersion() < int64(blockNumber) { //nolint:gosec // small test heights + time.Sleep(time.Millisecond) + } seed += uint64(receiptsPerBlock) if (block+1)%logInterval == 0 { diff --git a/sei-db/ledger_db/receipt/receipt_store.go b/sei-db/ledger_db/receipt/receipt_store.go index eea5219070..2a713c9881 100644 --- a/sei-db/ledger_db/receipt/receipt_store.go +++ b/sei-db/ledger_db/receipt/receipt_store.go @@ -51,12 +51,14 @@ func NewTooManyLogBytesError(maxBytes int64) error { type ReceiptStore interface { controller.PrunableStore + // LatestVersion is the highest block whose receipts are queryable. A write may land after + // SetReceipts returns, so a reader follows this rather than the height it last wrote. LatestVersion() int64 EarliestVersion() int64 - SetLatestVersion(version int64) error - SetEarliestVersion(version int64) error GetReceipt(ctx sdk.Context, txHash common.Hash) (*types.Receipt, error) GetReceiptFromStore(ctx sdk.Context, txHash common.Hash) (*types.Receipt, error) + // SetReceipts writes the block's receipts, carrying the version markers with them. An + // implementation may apply the write in the background; LatestVersion reports when it lands. SetReceipts(ctx sdk.Context, receipts []ReceiptRecord) error // FilterLogs queries logs across a range of blocks. // For single-block queries, set fromBlock == toBlock. @@ -68,6 +70,26 @@ type ReceiptStore interface { Close() error } +// VersionPinner is implemented by receipt stores whose version markers can be written directly. It +// is for a caller that put receipts in place by other means and has to state the window they cover. +type VersionPinner interface { + SetLatestVersion(version int64) error + SetEarliestVersion(version int64) error +} + +// PinVersions widens store's queryable window to [earliest, latest], reporting a store that cannot +// be pinned rather than leaving the window unset. +func PinVersions(store ReceiptStore, earliest, latest int64) error { + pinner, ok := store.(VersionPinner) + if !ok { + return fmt.Errorf("receipt store %T cannot pin versions", store) + } + if err := pinner.SetLatestVersion(latest); err != nil { + return err + } + return pinner.SetEarliestVersion(earliest) +} + type ReceiptRecord struct { TxHash common.Hash Receipt *types.Receipt diff --git a/sei-db/ledger_db/receipt/receipt_store_test.go b/sei-db/ledger_db/receipt/receipt_store_test.go index 28461dbed8..7d51966d53 100644 --- a/sei-db/ledger_db/receipt/receipt_store_test.go +++ b/sei-db/ledger_db/receipt/receipt_store_test.go @@ -93,6 +93,7 @@ func TestSetReceiptsAndGet(t *testing.T) { {TxHash: txHash}, }) require.NoError(t, err) + require.Eventually(t, func() bool { return store.LatestVersion() >= 1 }, 5*time.Second, time.Millisecond) got, err := store.GetReceipt(ctx, txHash) require.NoError(t, err) @@ -106,9 +107,10 @@ func TestSetReceiptsAndGet(t *testing.T) { require.Error(t, err) require.GreaterOrEqual(t, store.LatestVersion(), int64(1)) - require.NoError(t, store.SetLatestVersion(10)) + + // The version markers ride SetReceipts, so they are off the store's interface. + require.NoError(t, receipt.PinVersions(store, 1, 10)) require.Equal(t, int64(10), store.LatestVersion()) - require.NoError(t, store.SetEarliestVersion(1)) require.Equal(t, int64(1), store.EarliestVersion()) } diff --git a/sei-db/ledger_db/receipt/test_helpers_test.go b/sei-db/ledger_db/receipt/test_helpers_test.go index bbb8df60c8..09c1c89e18 100644 --- a/sei-db/ledger_db/receipt/test_helpers_test.go +++ b/sei-db/ledger_db/receipt/test_helpers_test.go @@ -1,13 +1,25 @@ package receipt import ( + "testing" + "time" + "github.com/ethereum/go-ethereum/common" storetypes "github.com/sei-protocol/sei-chain/sei-cosmos/store/types" "github.com/sei-protocol/sei-chain/sei-cosmos/testutil" sdk "github.com/sei-protocol/sei-chain/sei-cosmos/types" "github.com/sei-protocol/sei-chain/x/evm/types" + "github.com/stretchr/testify/require" ) +// requireReceiptVersion waits for the store to publish height. A write may be applied after +// SetReceipts returns, and LatestVersion is how a reader learns that it landed. +func requireReceiptVersion(t *testing.T, store ReceiptStore, height int64) { + t.Helper() + require.Eventually(t, func() bool { return store.LatestVersion() >= height }, + 5*time.Second, time.Millisecond) +} + func newTestContext() (sdk.Context, storetypes.StoreKey) { storeKey := storetypes.NewKVStoreKey("evm") tkey := storetypes.NewTransientStoreKey("evm_transient") diff --git a/sei-db/state_db/giga/state_db.go b/sei-db/state_db/giga/state_db.go index 8d7c224e35..036eb4e7d7 100644 --- a/sei-db/state_db/giga/state_db.go +++ b/sei-db/state_db/giga/state_db.go @@ -77,7 +77,7 @@ func NewStateDB( ) (db *StateDB, retErr error) { s := &StateDB{ flatkvCfg: flatkvCfg, - ssCfg: ssCfg, + ssCfg: stateStoreConfigFor(ssCfg), commitPhases: metrics.NewPhaseTimerFactory(otel.Meter(gigaMeterName), commitPhaseTimerName). RecordLatencies().Build(), } @@ -108,6 +108,14 @@ func NewStateDB( return s, nil } +// stateStoreConfigFor is the config a StateDB opens SS with. It is settled here rather than at each +// open because the rollback path opens the same databases through DiscardStateAbove. The changelog +// is off: this StateDB's own state WAL is what catchUpTo replays into SS. +func stateStoreConfigFor(cfg config.StateStoreConfig) config.StateStoreConfig { + cfg.DisableInternalWAL = true + return cfg +} + // NewStateDBWithRollback rolls SC, SS and the state WAL back to target and then opens them, so the // returned StateDB commits target+1. It cuts the WAL's tail to target and puts whichever of SC and SS // sits above target on its newest snapshot at or below it, all while the stores are closed, then opens @@ -131,7 +139,7 @@ func NewStateDBWithRollback( } // rewindTo only moves files, so it needs no store open, only where they live. - offline := &StateDB{flatkvCfg: flatkvCfg, ssCfg: ssCfg} + offline := &StateDB{flatkvCfg: flatkvCfg, ssCfg: stateStoreConfigFor(ssCfg)} if err := offline.rewindTo(target); err != nil { return nil, err } diff --git a/sei-db/state_db/giga/state_db_replay_test.go b/sei-db/state_db/giga/state_db_replay_test.go index f1861346c9..cfd41c9e8c 100644 --- a/sei-db/state_db/giga/state_db_replay_test.go +++ b/sei-db/state_db/giga/state_db_replay_test.go @@ -13,6 +13,57 @@ import ( "github.com/stretchr/testify/require" ) +// SS keeps no changelog of its own under giga, the state WAL being what catchUpTo replays into it. +// The absence is pinned here rather than left to the config, since recovery rests on it. +func TestGigaOpensSSWithoutAChangelog(t *testing.T) { + newStateDB := func(t *testing.T) *StateDB { + t.Helper() + ssCfg := config.DefaultStateStoreConfig() + ssCfg.Enable = true + ssCfg.EVMDBDirectory = filepath.Join(t.TempDir(), "ss") + return &StateDB{ + flatkvCfg: flatkvconfig.DefaultTestConfig(t), + // As the constructors settle it, which is what makes both paths below agree. + ssCfg: stateStoreConfigFor(ssCfg), + } + } + + t.Run("opened to commit", func(t *testing.T) { + s := newStateDB(t) + require.NoError(t, s.openSS()) + t.Cleanup(func() { _ = s.ss.Close() }) + requireNoSSChangelog(t, s.ssCfg.EVMDBDirectory) + }) + + // The rollback path opens the same databases through DiscardStateAbove rather than openSS, so a + // config settled per-open would miss it. StoredVersions opens nothing when the directory is + // absent, so the store has to exist first. + t.Run("opened to roll back", func(t *testing.T) { + s := newStateDB(t) + require.NoError(t, s.openSS()) + require.NoError(t, s.ss.Close()) + + require.NoError(t, s.discardStateAbove(storedWALRange{first: 1, last: 9}, 7)) + requireNoSSChangelog(t, s.ssCfg.EVMDBDirectory) + }) +} + +// TestStateStoreConfigForDisablesTheInternalWAL pins what the constructors apply, every path that +// opens SS reading the config they settled rather than disabling the log for itself. +func TestStateStoreConfigForDisablesTheInternalWAL(t *testing.T) { + handedIn := config.DefaultStateStoreConfig() + require.False(t, handedIn.DisableInternalWAL, "a caller is not expected to have set it") + require.True(t, stateStoreConfigFor(handedIn).DisableInternalWAL) +} + +func requireNoSSChangelog(t *testing.T, evmDBDirectory string) { + t.Helper() + changelog := utils.GetChangelogPath(evmDBDirectory) + _, err := os.Stat(changelog) + require.True(t, os.IsNotExist(err), + "SS must keep no changelog under giga; found one at %s", changelog) +} + // A node that keeps no EVM state store never reaches it, so nothing probes a store it does not have. // The directory is one an earlier run with SS on could have left, and the WAL reaches block 1, so a // rollback that read it would come back with a rewind to run.