diff --git a/.github/actions/setup-zeko-ci/action.yaml b/.github/actions/setup-zeko-ci/action.yaml new file mode 100644 index 0000000000..692975b01f --- /dev/null +++ b/.github/actions/setup-zeko-ci/action.yaml @@ -0,0 +1,93 @@ +name: Setup Zeko CI +description: Install the shared Zeko CI toolchain and restore dependency/build caches. +inputs: + opam-cache-key: + description: Cache key for the opam switch cache. + required: true + build-cache-key: + description: Cache key for the dune build cache. + required: true +runs: + using: composite + steps: + - name: ๐Ÿน Setup go + uses: actions/setup-go@v6 + with: + go-version: "1.19.11" + - name: ๐Ÿน Build and install capnpc-go + shell: bash + run: | + tmpdir=$(mktemp -d) + cd "$tmpdir" + go mod init local/build + go get capnproto.org/go/capnp/v3@v3.0.0-alpha.5 + go build -o capnpc-go capnproto.org/go/capnp/v3/capnpc-go + sudo cp capnpc-go /usr/local/bin/ + cd - + - uses: dtolnay/rust-toolchain@stable + - name: ๐Ÿ“ฅ Install OPAM + shell: bash + run: | + sudo apt-get update + sudo apt-get install -y opam ocaml \ + libboost-dev \ + libboost-program-options-dev \ + libbz2-dev \ + libcap-dev \ + libffi-dev \ + libgflags-dev \ + libgmp-dev \ + libgmp3-dev \ + libjemalloc-dev \ + liblmdb-dev \ + liblmdb0 \ + libpq-dev \ + libsodium-dev \ + libssl-dev \ + pkg-config zlib1g-dev \ + capnproto \ + netcat-traditional + for i in 1 2 3 4 5; do + rm -rf ./opam-repository + git clone https://github.com/ocaml/opam-repository.git --depth 1 ./opam-repository && break + echo "clone failed, retry $i" + getent hosts github.com || true + sleep 5 + done + git -C ./opam-repository fetch origin 08d8c16c16dc6b23a5278b06dff0ac6c7a217356 + git -C ./opam-repository checkout 08d8c16c16dc6b23a5278b06dff0ac6c7a217356 + opam init --disable-sandboxing -k git -a ./opam-repository --bare + for i in 1 2 3 4 5; do + opam repository add --yes --all --set-default o1-labs https://github.com/o1-labs/opam-repository.git && break + echo "opam repo add failed, retry $i" + sleep 5 + done + for i in 1 2 3 4 5; do + opam repository add --yes --all --set-default --rank=-1 opam-latest https://github.com/ocaml/opam-repository.git && break + echo "opam repo add failed, retry $i" + sleep 5 + done + opam switch create "4.14.0" + opam switch "4.14.0" + opam repository set-repos --yes --this-switch o1-labs default opam-latest + export OPAMERRLOGLEN=20 + export OPAMYES=1 + - name: ๐Ÿงฉ Get opam dependency cache + uses: WarpBuilds/cache@v1 + id: opam_dependencies + with: + path: ~/.opam + key: ${{ inputs.opam-cache-key }} + - name: ๐Ÿ—๏ธ Install opam dependencies + shell: bash + run: | + opam switch import opam.export --yes + eval "$(opam env)" + ./scripts/pin-external-packages.sh + opam clean --logs -cs --quiet + - name: ๐Ÿงฉ Get zeko build cache + uses: WarpBuilds/cache@v1 + id: zeko_build + with: + path: ./_build + key: ${{ inputs.build-cache-key }} diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index b00c4d41cf..0131b57654 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -17,87 +17,17 @@ jobs: uses: actions/checkout@v5 with: submodules: recursive - # Prepare build environment(Go/OCaml/Rust) - - name: ๐Ÿน Setup go - uses: actions/setup-go@v6 + - name: ๐Ÿงฐ Setup Zeko CI + uses: ./.github/actions/setup-zeko-ci + timeout-minutes: 20 with: - go-version: "1.19.11" - - name: ๐Ÿน Build and install capnpc-go - run: | - tmpdir=$(mktemp -d) - cd "$tmpdir" - go mod init local/build - go get capnproto.org/go/capnp/v3@v3.0.0-alpha.5 - go build -o capnpc-go capnproto.org/go/capnp/v3/capnpc-go - sudo cp capnpc-go /usr/local/bin/ - cd - - - uses: dtolnay/rust-toolchain@stable - # Install OPAM and needed dependencies - - name: ๐Ÿ“ฅ Install OPAM - run: | - sudo apt-get update - sudo apt-get install -y opam ocaml \ - libboost-dev \ - libboost-program-options-dev \ - libbz2-dev \ - libcap-dev \ - libffi-dev \ - libgflags-dev \ - libgmp-dev \ - libgmp3-dev \ - libjemalloc-dev \ - liblmdb-dev \ - liblmdb0 \ - libpq-dev \ - libsodium-dev \ - libssl-dev \ - pkg-config zlib1g-dev \ - capnproto \ - netcat-traditional - for i in 1 2 3 4 5; do - rm -rf ./opam-repository - git clone https://github.com/ocaml/opam-repository.git --depth 1 ./opam-repository && break - echo "clone failed, retry $i" - getent hosts github.com || true - sleep 5 - done - git -C ./opam-repository fetch origin 08d8c16c16dc6b23a5278b06dff0ac6c7a217356 - git -C ./opam-repository checkout 08d8c16c16dc6b23a5278b06dff0ac6c7a217356 - opam init --disable-sandboxing -k git -a ./opam-repository --bare - for i in 1 2 3 4 5; do - opam repository add --yes --all --set-default o1-labs https://github.com/o1-labs/opam-repository.git && break - echo "opam repo add failed, retry $i" - sleep 5 - done - opam switch create "4.14.0" - opam switch "4.14.0" - export OPAMERRLOGLEN=20 - export OPAMYES=1 - # Retrieve .opam from cache - - name: ๐Ÿงฉ Get opam dependency cache - uses: WarpBuilds/cache@v1 - id: opam_dependencies - with: - path: ~/.opam - key: zeko_opam_dependencies-${{ hashFiles('./opam.export') }} - # Install opam dependencies - - name: ๐Ÿ—๏ธ Install opam dependencies - run: | - opam switch import opam.export --yes - eval "$(opam env)" - ./scripts/pin-external-packages.sh - opam clean --logs -cs --quiet - # Retrieve _build dir from cache before initiating a build - - name: ๐Ÿงฉ Get zeko build cache - uses: WarpBuilds/cache@v1 - id: zeko_build - with: - path: ./_build - key: zeko_build-${{ hashFiles('./opam.export') }} + opam-cache-key: zeko_opam_dependencies-${{ hashFiles('./opam.export') }} + build-cache-key: zeko_build-${{ hashFiles('./opam.export') }} # Build! - name: ๐Ÿ“ฆ Build zeko packages run: | opam exec -- dune build --profile=devnet src/app/zeko/sequencer + opam exec -- dune build --profile=devnet src/app/zeko/sequencer/explorer opam exec -- dune build --profile=devnet src/app/zeko/da_layer opam exec -- dune build --profile=devnet src/app/zeko/signer - name: ๐Ÿงช Run sequencer tests diff --git a/.github/workflows/explorer-e2e.yaml b/.github/workflows/explorer-e2e.yaml new file mode 100644 index 0000000000..7fb569edac --- /dev/null +++ b/.github/workflows/explorer-e2e.yaml @@ -0,0 +1,45 @@ +# Runs the slow explorer end-to-end Gherkin scenarios independently from the +# faster explorer integration suite and the existing sequencer integration run. +--- +name: Explorer E2E + +on: + push: + +concurrency: + group: "${{ github.workflow }} @ ${{ github.head_ref || github.ref }}" + cancel-in-progress: true + +jobs: + explorer-e2e: + name: "Explorer E2E" + runs-on: warp-ubuntu-latest-x64-16x + services: + nats: + image: nats:2-alpine + command: -js + ports: + - 4222:4222 + steps: + - name: ๐Ÿ“ฅ Checkout + uses: actions/checkout@v5 + with: + submodules: recursive + - name: ๐Ÿงฐ Setup Zeko CI + uses: ./.github/actions/setup-zeko-ci + timeout-minutes: 20 + with: + opam-cache-key: zeko_opam_dependencies-${{ hashFiles('./opam.export') }} + build-cache-key: zeko_build-${{ hashFiles('./opam.export') }} + - name: ๐Ÿงช Run explorer E2E Gherkin tests + run: | + for i in $(seq 1 30); do + nc -z 127.0.0.1 4222 && break + if [ "$i" -eq 30 ]; then + echo "Timed out waiting for NATS" >&2 + exit 1 + fi + sleep 1 + done + NATS_URL="nats://127.0.0.1:4222" \ + ./src/app/zeko/sequencer/explorer/tests/run-explorer-e2e-test.sh real 1 diff --git a/.github/workflows/explorer-gherkin.yaml b/.github/workflows/explorer-gherkin.yaml new file mode 100644 index 0000000000..025a2eaf2f --- /dev/null +++ b/.github/workflows/explorer-gherkin.yaml @@ -0,0 +1,46 @@ +# Runs the explorer Gherkin acceptance tests independently from the sequencer +# integration suite so PR feedback is not blocked on the longer sequencer flow. +--- +name: Explorer Gherkin + +on: + push: + +concurrency: + group: "${{ github.workflow }} @ ${{ github.head_ref || github.ref }}" + cancel-in-progress: true + +jobs: + explorer-gherkin: + name: "Explorer Gherkin" + runs-on: warp-ubuntu-latest-x64-16x + services: + nats: + image: nats:2-alpine + command: -js + ports: + - 4222:4222 + steps: + - name: ๐Ÿ“ฅ Checkout + uses: actions/checkout@v5 + with: + submodules: recursive + - name: ๐Ÿงฐ Setup Zeko CI + uses: ./.github/actions/setup-zeko-ci + timeout-minutes: 20 + with: + opam-cache-key: zeko_opam_dependencies-${{ hashFiles('./opam.export') }} + build-cache-key: zeko_build-${{ hashFiles('./opam.export') }} + - name: ๐Ÿงช Run explorer Gherkin tests + run: | + for i in $(seq 1 30); do + nc -z 127.0.0.1 4222 && break + if [ "$i" -eq 30 ]; then + echo "Timed out waiting for NATS" >&2 + exit 1 + fi + sleep 1 + done + NATS_URL="nats://127.0.0.1:4222" \ + opam exec -- env -u DUNE_RPC dune runtest --profile=devnet \ + src/app/zeko/sequencer/explorer/tests diff --git a/.prototools b/.prototools new file mode 100644 index 0000000000..4ec55a8388 --- /dev/null +++ b/.prototools @@ -0,0 +1,4 @@ +ocaml = "4.14.0" + +[plugins.tools] +ocaml = "github://Hebilicious/proto-ocaml@v0.1.2" diff --git a/opam.export b/opam.export index bdb02a27ea..5bdc186446 100644 --- a/opam.export +++ b/opam.export @@ -30,6 +30,8 @@ roots: [ "lmdb.1.0" "menhir.20210419" "merlin.4.5-414" + "nats-client.0.0.10" + "nats-client-async.0.0.10" "ocaml-base-compiler.4.14.0" "ocamlformat.0.20.1" "ocamlgraph.1.8.8" @@ -187,6 +189,8 @@ installed: [ "mirage-crypto-rng.0.11.0" "mirage-crypto-rng-async.0.11.0" "mmap.1.1.0" + "nats-client.0.0.10" + "nats-client-async.0.0.10" "num.1.1" "ocaml.4.14.0" "ocaml-base-compiler.4.14.0" diff --git a/src/app/zeko/sequencer/README.md b/src/app/zeko/sequencer/README.md index b847cf2d4b..b3ff54a973 100644 --- a/src/app/zeko/sequencer/README.md +++ b/src/app/zeko/sequencer/README.md @@ -45,16 +45,68 @@ dune exec ./run.exe -- \ --deposit-delay-blocks \ --fee-modifier \ --minimum-fee \ + --nats-url \ --slot-acceptance \ --commit-validity-period ``` +`--nats-url` enables explorer event publishing. Without it, the sequencer keeps +the existing behavior and all NATS publish paths become no-ops. + +When NATS is enabled, the sequencer emits these NATS subjects: + +- `zeko.l2.transactions` +- `zeko.l2.finality` +- `zeko.health` + +The exact subjects, JetStream streams, payload fields, and dedup headers are +documented in `src/app/zeko/sequencer/explorer/MESSAGE_CONTRACT.md`. + Run help to see the options: ```bash dune exec ./run.exe -- --help ``` +## Explorer backfill service + +Build it with: + +```bash +DUNE_PROFILE=devnet dune build ./src/app/zeko/sequencer/explorer +``` + +Run it with: + +```bash +export DUNE_PROFILE=devnet +dune exec ./explorer/explorer_backfill_server.exe -- \ + -p \ + --da-node \ + --nats-url +``` + +The backfill API is a separate process from the sequencer GraphQL API and +republishes historical DA diffs into `zeko.l2.transactions`. It requires +`--nats-url` because every backfill job publishes to JetStream. + +Available GraphQL operations: + +- `backfill(fromHash, toHash) -> BackfillJob` +- `backfillJob(id) -> BackfillJob | null` +- `health -> Health` + +GraphQL-SSE endpoint: + +- `backfillProgress(id) -> BackfillProgress` + +## Explorer acceptance tests + +Explorer-specific Gherkin tests live under +`src/app/zeko/sequencer/explorer/tests/`. See +`src/app/zeko/sequencer/explorer/tests/README.md` for the fast integration, +NATS-backed, and E2E commands. + ## Deploy rollup contract to L1 The following script deploys the rollup contract on the L1 with the initial state, which is the genesis ledger of the rollup. diff --git a/src/app/zeko/sequencer/explorer/MESSAGE_CONTRACT.md b/src/app/zeko/sequencer/explorer/MESSAGE_CONTRACT.md new file mode 100644 index 0000000000..4a9770e423 --- /dev/null +++ b/src/app/zeko/sequencer/explorer/MESSAGE_CONTRACT.md @@ -0,0 +1,157 @@ +# Zeko Explorer JetStream Message Contract + +This document is the Zeko-owned contract for explorer-facing NATS/JetStream +publishing from the sequencer and backfill service. + +## Streams + +- Stream name: `zeko-l2` + - Subjects: `zeko.l2.>` + - Duplicate window: 120 seconds + - Max age: 90 days + - Storage: file + - Retention: limits +- Stream name: `zeko-health` + - Subjects: `zeko.health` + - Max messages: 1000 + - Storage: file + - Retention: limits + +Sequencer startup best-effort ensures the streams exist with these subjects when +`--nats-url` is configured. If NATS is not configured, sequencer publishing is a +no-op. If NATS is configured but unavailable, the sequencer logs a warning and +explorer publishing remains disabled rather than crashing. + +The standalone backfill service is stricter: `--nats-url` is required, startup +fails if it cannot connect to JetStream, and dropped backfill publishes fail the +backfill job. + +## Transaction Diffs + +Subject: `zeko.l2.transactions` + +Headers: + +```text +Nats-Msg-Id: +``` + +`target_ledger_hash_decimal` is `Ledger_hash.to_decimal_string +target_ledger_hash`. Publishing the same transaction diff with the same target +ledger hash must deduplicate at the JetStream stream level. + +Payload: + +```json +{ + "kind": "user_command | fee_transfer | sync_replay | genesis_replay", + "source_ledger_hash": "", + "target_ledger_hash": "", + "timestamp": "", + "acc_set": "", + "command": { + "type": "signed_command | zkapp_command", + "raw": "", + "action_step_flags": [true] + }, + "changed_accounts": [ + { + "index": 0, + "account": "" + } + ], + "command_with_action_step_flags": "", + "actions": "", + "genesis": false, + "diff": { + "...": "Da_layer.Diff.Stable.V4.to_yojson diff" + } +} +``` + +Fields: + +- `kind`: why this diff is being published. +- `source_ledger_hash`: source ledger hash encoded with + `Ledger_hash.to_yojson`. +- `target_ledger_hash`: target ledger hash encoded with + `Ledger_hash.to_yojson`. +- `timestamp`: diff timestamp encoded with `Block_time.to_yojson`. +- `acc_set`: account-set root from the DA diff. +- `command`: normalized command envelope. `null` for fee transfers and diffs + without a command. +- `changed_accounts`: post-state accounts affected by the diff, normalized as + `{index, account}` objects. +- `command_with_action_step_flags`: legacy command tuple projection, retained for + consumers that still read the previous command field. +- `actions`: raw DA-layer actions field from the current diff format. +- `genesis`: `true` when replaying the first diff from the genesis source hash. +- `diff`: DA-layer diff encoded with `Da_layer.Diff.Stable.V4.to_yojson`. + +## Finality + +Subject: `zeko.l2.finality` + +Headers: + +```text +Nats-Msg-Id: finality-- +``` + +`status` is the payload status string, and `target_ledger_hash_decimal` is +`Ledger_hash.to_decimal_string target_ledger_hash`. This keeps `proved` and +`committed` finality messages independently deduplicated for the same target +ledger hash. + +Payload: + +```json +{ + "status": "proved | committed", + "level": "proved | committed", + "ledger_hash": "", + "source_ledger_hash": "", + "target_ledger_hash": "", + "timestamp": "" +} +``` + +Fields: + +- `level`: finality state for the target ledger hash. +- `ledger_hash`: target ledger hash encoded with `Ledger_hash.to_yojson`. +- `status`: alias of `level`, retained for compatibility with existing + Zeko-side consumers. +- `source_ledger_hash`: source ledger hash encoded with + `Ledger_hash.to_yojson`. +- `target_ledger_hash`: target ledger hash encoded with + `Ledger_hash.to_yojson`. +- `timestamp`: publisher timestamp encoded with `Block_time.to_yojson`. + +## Health + +Subject: `zeko.health` + +Headers: none required. + +Payload: + +```json +{ + "service": "sequencer-nats-publisher", + "instance_id": "", + "status": "ok", + "last_published_hash": "", + "unproved_hash": "", + "timestamp": "" +} +``` + +Fields: + +- `service`: publishing service identity. +- `instance_id`: publisher instance identity. +- `status`: publisher health/status string. +- `last_published_hash`: optional last published ledger hash. +- `unproved_hash`: optional unproved ledger hash. +- `timestamp`: publisher timestamp encoded with `Block_time.to_yojson`. diff --git a/src/app/zeko/sequencer/explorer/dune b/src/app/zeko/sequencer/explorer/dune new file mode 100644 index 0000000000..037a9d4e35 --- /dev/null +++ b/src/app/zeko/sequencer/explorer/dune @@ -0,0 +1,75 @@ +; Explorer support libraries and the standalone backfill server executable. + +(library + (name explorer_events) + (modules explorer_events) + (libraries + da_layer + nats-client + nats-client-async + ;; mina ;; + mina_base + snark_params + ;; opam libraries ;; + yojson + uri + core_kernel + async + ppx_inline_test.config) + (preprocess + (pps ppx_jane ppx_mina))) + +(library + (name explorer_backfill) + (wrapped false) + (modules + explorer_backfill_service + explorer_backfill_graphql + explorer_backfill_sse) + (libraries + explorer_events + sequencer_lib + da_layer + zeko_constants + init + cli_lib + nats-client + nats-client-async + ;; mina ;; + mina_base + snark_params + ;; opam libraries ;; + cohttp + cohttp-async + graphql + graphql-async + uri + core + core_kernel + async + async_unix + ppx_inline_test.config) + (preprocess + (pps ppx_let ppx_jane ppx_mina))) + +(executable + (name explorer_backfill_server) + (modules explorer_backfill_server) + (libraries + explorer_backfill + sequencer_lib + is_compile_simple_real_real + init + cli_lib + ;; opam libraries ;; + cohttp + cohttp-async + graphql + graphql-async + core + core_kernel + async + async_unix + core_unix.command_unix) + (preprocess + (pps ppx_let ppx_jane ppx_mina))) diff --git a/src/app/zeko/sequencer/explorer/explorer-indexing-plan.md b/src/app/zeko/sequencer/explorer/explorer-indexing-plan.md new file mode 100644 index 0000000000..d2f828f549 --- /dev/null +++ b/src/app/zeko/sequencer/explorer/explorer-indexing-plan.md @@ -0,0 +1,334 @@ +# Explorer Indexing Plan + +## High Level + +This work adds a first-party NATS publishing path to the Zeko sequencer and a standalone backfill service in this repo. + +When it is done, this repo will be able to: + +- publish L2 transaction events as the sequencer processes diffs +- publish L2 finality events as batches move from proved to committed +- publish health heartbeats for freshness monitoring +- republish historical diff ranges into NATS through a standalone backfill service + +This matters because explorer and indexing systems need a direct, replayable event stream from the sequencer and DA layer, without depending on downstream reconstruction of sequencer state. + +## 1. Integrate `nats-ml` into this repo + +Work: + +- Bring `[nats-ml](https://github.com/Hebilicious/nats-ml)` into this repo from GitHub. +- Wire it into the repo build so sequencer code can depend on it. +- Make it available to both the sequencer and the backfill service. + +Result: + +- NATS publishing depends on a repo-local integration of `nats-ml`, not an external manual install step. + +## 2. Add sequencer NATS configuration + +Target files: + +- `src/app/zeko/sequencer/run.ml` +- `src/app/zeko/sequencer/lib/zeko_sequencer.ml` +- `src/app/zeko/sequencer/dune` + +Work: + +- Add `--nats-url` to the sequencer CLI. +- Extend sequencer config and runtime state with an optional NATS client. +- Create the `nats-ml` client at startup when `--nats-url` is present. +- Keep all publishing paths as no-ops when NATS is not configured. + +Result: + +- The sequencer runs unchanged without NATS. +- The sequencer publishes to NATS when configured. + +## 3. Add one shared NATS event module + +Target files: + +- `src/app/zeko/sequencer/lib/` + +Work: + +- Add a small module that owns NATS subjects, headers, and JSON encoding. +- Use this module from both live sequencer publishing and the backfill service. +- Keep the event contract in one place. + +Subjects: + +- `zeko.l2.transactions` +- `zeko.l2.finality` +- `zeko.health` + +Headers: + +- `Nats-Msg-Id: ` on every `zeko.l2.transactions` message + +### `zeko.l2.transactions` schema + +Fields: + +- `kind` + - `user_command` + - `fee_transfer` + - `sync_replay` + - `genesis_replay` +- `source_ledger_hash` +- `target_ledger_hash` +- `timestamp` +- `acc_set` +- `command` +- `changed_accounts` +- `command_with_action_step_flags` +- `genesis` +- `diff` + +Encoding rules: + +- `source_ledger_hash`, `timestamp`, `acc_set`, and + `actions` come from the JSON shape derived from + `Da_layer.Diff.Stable.V4`. +- `command` is a normalized envelope with `type`, `raw`, and + `action_step_flags`, or `null` when the diff has no command. +- `changed_accounts` is normalized to `{index, account}` objects. +- `diff` retains the existing JSON shape derived from + `Da_layer.Diff.Stable.V4`. +- `target_ledger_hash` is the post-diff ledger hash passed to `Da_layer.Client.enqueue_diff`. +- `genesis` matches the flag passed to `Da_layer.Client.enqueue_diff`. + +### `zeko.l2.finality` schema + +Fields: + +- `level` + - `proved` + - `committed` +- `ledger_hash` +- `source_ledger_hash` +- `target_ledger_hash` +- `status` +- `timestamp` + +Encoding rules: + +- `proved` is emitted from the successful committer result. +- `committed` is emitted after `State.Last_committed_ledger.set`. +- `status` is retained as an alias of `level`. + +### `zeko.health` schema + +Fields: + +- `service` +- `instance_id` +- `status` +- `last_published_hash` +- `unproved_hash` +- `timestamp` + +Encoding rules: + +- `service = "sequencer-nats-publisher"` +- `status = "ok"` + +Result: + +- This repo has one stable event boundary for explorer ingestion. + +## 4. Publish L2 transaction events from the sequencer + +Target file: + +- `src/app/zeko/sequencer/lib/zeko_sequencer.ml` + +Integration points: + +- `apply_user_command` +- `apply_fee_transfer` +- `sync` + +Work: + +- After `Da_layer.Client.enqueue_diff` in `apply_user_command`, publish one `zeko.l2.transactions` message with `kind = user_command`. +- After `Da_layer.Client.enqueue_diff` in `apply_fee_transfer`, publish one `zeko.l2.transactions` message with `kind = fee_transfer`. +- During `sync`, publish replayed diffs after they are re-enqueued. +- Use `kind = genesis_replay` when the replayed diff was enqueued with `genesis = true`. +- Use `kind = sync_replay` for the rest of the replayed diffs. +- Set `Nats-Msg-Id` from `target_ledger_hash`. + +Result: + +- Live traffic and replayed traffic produce the same transaction event format. + +## 5. Publish finality events from the sequencer + +Target file: + +- `src/app/zeko/sequencer/lib/zeko_sequencer.ml` + +Integration points: + +- `run_committer` +- `Merger.Commit.process` + +Work: + +- Publish `zeko.l2.finality` with `status = proved` when `run_committer` receives `Some (stmt, _)`. +- Publish `zeko.l2.finality` with `status = committed` after `State.Last_committed_ledger.set` in the commit path. +- Include both source and target ledger hashes in both cases. + +Result: + +- The explorer stack can track proved and committed progress directly from NATS. + +## 6. Publish sequencer health heartbeats + +Target file: + +- `src/app/zeko/sequencer/lib/zeko_sequencer.ml` + +Work: + +- Add a periodic heartbeat loop. +- Emit `zeko.health` messages using the shared event module. +- Generate one sequencer instance id at startup and reuse it for all heartbeats from that process. + +Result: + +- The sequencer emits a freshness signal even when no transactions are being processed. + +## 7. Add the backfill service + +Target area: + +- new standalone executable under `src/app/zeko/sequencer/` +- build wiring in `src/app/zeko/sequencer/dune` + +Work: + +- Add a standalone OCaml service that reads diffs from the DA layer and republishes them to NATS. +- Use `nats-ml` for publishing. +- Reuse the shared NATS event module so backfilled messages match live messages exactly. +- Track backfill jobs in memory. +- Generate one service instance id at startup. + +GraphQL API: + +- `backfill(fromHash, toHash) -> BackfillJob` +- `backfillJob(id) -> BackfillJob` +- `health -> Health` + +GraphQL-SSE subscription: + +- `backfillProgress(id) -> BackfillProgress` + +`BackfillJob` fields: + +- `id` +- `fromHash` +- `toHash` +- `status` +- `diffsPublished` +- `error` +- `createdAt` +- `startedAt` +- `finishedAt` + +`BackfillProgress` fields: + +- `id` +- `status` +- `diffsPublished` +- `error` + +`Health` fields: + +- `ok` +- `instanceId` +- `startedAt` + +Transport: + +- Use GraphQL over HTTP for query and mutation. +- Use GraphQL-SSE for `backfillProgress`. + +Publishing rules: + +- Republish to `zeko.l2.transactions`. +- Use the same payload schema and `Nats-Msg-Id` rule as the live sequencer path. +- Emit `kind = genesis_replay` only when replaying a genesis diff. +- Emit `kind = sync_replay` for all other backfilled diffs. + +Result: + +- Missing L2 ranges can be republished into NATS without changing sequencer state. + +## 8. Update sequencer docs + +Target file: + +- `src/app/zeko/sequencer/README.md` + +Work: + +- Document `--nats-url`. +- Document the three emitted subjects. +- Document the exact payload families. +- Document how to run the backfill service. +- Document the backfill GraphQL and GraphQL-SSE surface. + +Result: + +- The repo docs describe the explorer-facing ingestion path end to end. + +## 9. Add tests as first-class work + +Work: + +- Add tests for the shared event module: + - transaction payload encoding + - finality payload encoding + - health payload encoding + - `Nats-Msg-Id` generation +- Add sequencer publishing tests for: + - `apply_user_command` + - `apply_fee_transfer` + - `sync` replay + - `run_committer` proved event + - commit-path committed event + - health heartbeat emission +- Add backfill service tests for: + - `backfill` job creation + - ordered republishing across a hash range + - progress updates over GraphQL-SSE + - `health` response contents + - matching payload shape between live and backfilled messages + +Result: + +- The explorer-facing boundary is covered directly in this repo. + +## Delivery Order + +1. Integrate `nats-ml` into this repo. +2. Add sequencer startup/config wiring for NATS. +3. Add the shared NATS event module. +4. Publish `zeko.l2.transactions`. +5. Publish `zeko.l2.finality`. +6. Publish `zeko.health`. +7. Add the backfill service with GraphQL and GraphQL-SSE. +8. Finish docs and tests. + +## Deliverable + +After this work, the `zeko` repo provides: + +- repo-local `nats-ml` integration +- optional sequencer publishing to NATS +- replay of historical diffs during sync +- finality and health events +- a standalone backfill service with GraphQL and GraphQL-SSE +- test coverage around the explorer-facing publish boundary diff --git a/src/app/zeko/sequencer/explorer/explorer_backfill_graphql.ml b/src/app/zeko/sequencer/explorer/explorer_backfill_graphql.ml new file mode 100644 index 0000000000..d43fa3eaea --- /dev/null +++ b/src/app/zeko/sequencer/explorer/explorer_backfill_graphql.ml @@ -0,0 +1,130 @@ +(* Defines the GraphQL schema for the standalone backfill server on top of the + Explorer_backfill_service job and health snapshots. *) + +open Core +open Async + +open Graphql_async.Schema + +let backfill_job_typ : + (Explorer_backfill_service.t, Explorer_backfill_service.job_snapshot option) typ + = + obj "BackfillJob" + ~fields:(fun _ -> + [ field "id" ~typ:(non_null string) ~args:[] + ~resolve:(fun _ (job : Explorer_backfill_service.job_snapshot) -> job.id) + ; field "fromHash" ~typ:(non_null string) ~args:[] + ~resolve:(fun _ (job : Explorer_backfill_service.job_snapshot) -> + job.from_hash) + ; field "toHash" ~typ:(non_null string) ~args:[] + ~resolve:(fun _ (job : Explorer_backfill_service.job_snapshot) -> + job.to_hash) + ; field "status" ~typ:(non_null string) ~args:[] + ~resolve:(fun _ (job : Explorer_backfill_service.job_snapshot) -> + job.status) + ; field "diffsPublished" ~typ:(non_null int) ~args:[] + ~resolve:(fun _ (job : Explorer_backfill_service.job_snapshot) -> + job.diffs_published) + ; field "error" ~typ:string ~args:[] + ~resolve:(fun _ (job : Explorer_backfill_service.job_snapshot) -> + job.error) + ; field "createdAt" ~typ:(non_null string) ~args:[] + ~resolve:(fun _ (job : Explorer_backfill_service.job_snapshot) -> + job.created_at) + ; field "startedAt" ~typ:string ~args:[] + ~resolve:(fun _ (job : Explorer_backfill_service.job_snapshot) -> + job.started_at) + ; field "finishedAt" ~typ:string ~args:[] + ~resolve:(fun _ (job : Explorer_backfill_service.job_snapshot) -> + job.finished_at) + ] ) + +let progress_typ : + (Explorer_backfill_service.t, Explorer_backfill_service.progress_snapshot option) typ + = + obj "BackfillProgress" + ~fields:(fun _ -> + [ field "id" ~typ:(non_null string) ~args:[] + ~resolve: + (fun _ (progress : Explorer_backfill_service.progress_snapshot) -> + progress.id) + ; field "status" ~typ:(non_null string) ~args:[] + ~resolve: + (fun _ (progress : Explorer_backfill_service.progress_snapshot) -> + progress.status) + ; field "diffsPublished" ~typ:(non_null int) ~args:[] + ~resolve: + (fun _ (progress : Explorer_backfill_service.progress_snapshot) -> + progress.diffs_published) + ; field "error" ~typ:string ~args:[] + ~resolve: + (fun _ (progress : Explorer_backfill_service.progress_snapshot) -> + progress.error) + ] ) + +let health_typ : + (Explorer_backfill_service.t, Explorer_backfill_service.health_snapshot option) typ + = + obj "Health" + ~fields:(fun _ -> + [ field "ok" ~typ:(non_null bool) ~args:[] + ~resolve:(fun _ (value : Explorer_backfill_service.health_snapshot) -> + value.ok) + ; field "instanceId" ~typ:(non_null string) ~args:[] + ~resolve:(fun _ (value : Explorer_backfill_service.health_snapshot) -> + value.instance_id) + ; field "startedAt" ~typ:(non_null string) ~args:[] + ~resolve:(fun _ (value : Explorer_backfill_service.health_snapshot) -> + value.started_at) + ] ) + +let query_fields = + [ io_field "backfillJob" ~typ:backfill_job_typ + ~args:Arg.[ arg "id" ~typ:(non_null string) ] + ~resolve:(fun { ctx; _ } () id -> + return + (Ok + (Option.map (Explorer_backfill_service.find_job ctx id) + ~f:Explorer_backfill_service.snapshot ) ) ) + ; io_field "health" ~typ:(non_null health_typ) ~args:[] + ~resolve:(fun { ctx; _ } () -> + return (Ok (Explorer_backfill_service.health ctx))) + ] + +let mutation_fields = + [ io_field "backfill" ~typ:(non_null backfill_job_typ) + ~args: + Arg. + [ arg "fromHash" ~typ:(non_null string) + ; arg "toHash" ~typ:(non_null string) + ] + ~resolve:(fun { ctx; _ } () from_hash to_hash -> + let snapshot = + match + Explorer_backfill_service.start_backfill_from_strings ctx ~from_hash + ~to_hash + with + | Ok job -> + Explorer_backfill_service.snapshot job + | Error err -> + Explorer_backfill_service.failed_job_snapshot_from_strings + ~from_hash ~to_hash (Error.to_string_hum err) + in + return (Ok snapshot)) + ] + +let subscription_fields = + [ subscription_field "backfillProgress" ~typ:(non_null progress_typ) + ~args:Arg.[ arg "id" ~typ:(non_null string) ] + ~resolve:(fun { ctx; _ } id -> + match Explorer_backfill_service.subscribe_progress ctx ~id with + | Ok progress -> + Deferred.Result.return progress + | Error err -> + Deferred.return (Error (Error.to_string_hum err))) + ] + +let schema = + Graphql_async.Schema.( + schema query_fields ~mutations:mutation_fields + ~subscriptions:subscription_fields) diff --git a/src/app/zeko/sequencer/explorer/explorer_backfill_server.ml b/src/app/zeko/sequencer/explorer/explorer_backfill_server.ml new file mode 100644 index 0000000000..5b71257528 --- /dev/null +++ b/src/app/zeko/sequencer/explorer/explorer_backfill_server.ml @@ -0,0 +1,66 @@ +(* Boots the standalone explorer backfill HTTP server and routes GraphQL query + traffic plus GraphQL-SSE subscription traffic to the backfill modules. *) + +open Core +open Async +open Cli_lib + +module Graphql_cohttp_async = + Init.Graphql_internal.Make (Graphql_async.Schema) (Cohttp_async.Io) + (Cohttp_async.Body) + +let run ~logger ~port ~da_config ~nats_url () = + let service = + Thread_safe.block_on_async_exn (fun () -> + Explorer_backfill_service.create ~logger ~da_config ~nats_url ) + in + let graphql_callback = + Graphql_cohttp_async.make_callback + (fun ~with_seq_no:_ _req -> service) + Explorer_backfill_graphql.schema + in + let callback ~body _sock req = + match Uri.path (Cohttp.Request.uri req) with + | "/graphql/stream" -> + Explorer_backfill_sse.callback service () req body + | _ -> + graphql_callback () req body + in + let () = + Cohttp_async.Server.create_expert + ~on_handler_error: + (`Call + (fun _ exn -> + [%log error] "Unhandled exception: %s" (Exn.to_string exn) ) ) + (Async.Tcp.Where_to_listen.of_port port) + callback + |> Deferred.ignore_m |> don't_wait_for + in + Shutdown.at_shutdown (fun () -> Explorer_backfill_service.shutdown service) ; + [%log info] "Explorer backfill server listening on port %d" port ; + never_returns (Async.Scheduler.go ()) + +let () = + Command.basic ~summary:"Zeko explorer backfill server" + (let%map_open.Command log_json = Flag.Log.json + and log_level = Flag.Log.level + and port = + flag "-p" + (optional_with_default 8090 int) + ~doc: + "int Port for the standalone backfill GraphQL API; keep distinct \ + from the sequencer GraphQL port when both run on the same host" + and da_nodes = + flag "--da-node" (listed string) + ~doc:"string Address of the DA node, can be supplied multiple times" + and nats_url = + flag "--nats-url" (required string) + ~doc:"string NATS URL for republishing explorer events" + in + if List.is_empty da_nodes then failwith "--da-node must be supplied" ; + Stdout_log.setup log_json log_level ; + let logger = Logger.create () in + let da_config = Da_layer.Client.Config.of_string_list da_nodes in + let nats_url = Uri.of_string nats_url in + run ~logger ~port ~da_config ~nats_url () ) + |> Command_unix.run diff --git a/src/app/zeko/sequencer/explorer/explorer_backfill_service.ml b/src/app/zeko/sequencer/explorer/explorer_backfill_service.ml new file mode 100644 index 0000000000..5e3540980c --- /dev/null +++ b/src/app/zeko/sequencer/explorer/explorer_backfill_service.ml @@ -0,0 +1,530 @@ +(* Owns explorer backfill job state and execution: tracks in-memory jobs, + republishes DA diffs to NATS, and exposes the snapshots used by the GraphQL + and SSE transport layers. *) + +open Core +open Async +open Mina_base + +type job_status = + | Queued + | Running + | Completed + | Failed + +type job_snapshot = + { id : string + ; from_hash : string + ; to_hash : string + ; status : string + ; diffs_published : int + ; error : string option + ; created_at : string + ; started_at : string option + ; finished_at : string option + } + +type progress_snapshot = + { id : string + ; status : string + ; diffs_published : int + ; error : string option + } + +type subscriber = + { writer : progress_snapshot Pipe.Writer.t + ; mutable pending : progress_snapshot option + ; mutable draining : bool + ; mutable close_after_flush : bool + } + +type health_snapshot = + { ok : bool + ; instance_id : string + ; started_at : string + } + +type job = + { id : string + ; from_hash : Ledger_hash.t + ; to_hash : Ledger_hash.t + ; mutable status : job_status + ; mutable diffs_published : int + ; mutable error : string option + ; created_at : Time.t + ; mutable started_at : Time.t option + ; mutable finished_at : Time.t option + ; subscribers : subscriber list ref + } + +type t = + { logger : Logger.t + ; da_config : Da_layer.Client.Config.t + ; nats_client : Nats_client_async.client option + ; jobs : job String.Table.t + ; instance_id : string + ; started_at : Time.t + } + +let constraint_constants = Zeko_constants.constraint_constants + +let timestamp_string time = + Time.to_string_iso8601_basic time ~zone:Time.Zone.utc + +let string_of_job_status = function + | Queued -> + "queued" + | Running -> + "running" + | Completed -> + "completed" + | Failed -> + "failed" + +let is_terminal = function + | Completed | Failed -> + true + | Queued | Running -> + false + +let terminal_job_retention = Time.Span.of_hr 1. + +let backfill_interval_size = 1000 + +let max_active_backfill_jobs = 1 + +let genesis_hash = + Da_layer.Diff.empty_ledger_hash ~depth:constraint_constants.ledger_depth + +let is_genesis_hash ledger_hash = Ledger_hash.equal ledger_hash genesis_hash + +let snapshot job = + { id = job.id + ; from_hash = Ledger_hash.to_decimal_string job.from_hash + ; to_hash = Ledger_hash.to_decimal_string job.to_hash + ; status = string_of_job_status job.status + ; diffs_published = job.diffs_published + ; error = job.error + ; created_at = timestamp_string job.created_at + ; started_at = Option.map job.started_at ~f:timestamp_string + ; finished_at = Option.map job.finished_at ~f:timestamp_string + } + +let progress_of_job job = + { id = job.id + ; status = string_of_job_status job.status + ; diffs_published = job.diffs_published + ; error = job.error + } + +let health t = + { ok = true + ; instance_id = t.instance_id + ; started_at = timestamp_string t.started_at + } + +let rec flush_subscriber subscriber = + let open Deferred.Let_syntax in + match subscriber.pending with + | None -> + subscriber.draining <- false ; + if subscriber.close_after_flush then Pipe.close subscriber.writer ; + Deferred.unit + | Some progress -> + subscriber.pending <- None ; + let%bind () = Pipe.write_if_open subscriber.writer progress in + flush_subscriber subscriber + +let create_subscriber writer = + { writer; pending = None; draining = false; close_after_flush = false } + +let enqueue_progress subscriber progress ~terminal = + if Pipe.is_closed subscriber.writer then false + else ( + subscriber.pending <- Some progress ; + if terminal then subscriber.close_after_flush <- true ; + if not subscriber.draining then ( + subscriber.draining <- true ; + don't_wait_for (flush_subscriber subscriber) ) ; + true ) + +let notify_subscribers job = + let progress = progress_of_job job in + let terminal = is_terminal job.status in + let subscribers = + List.filter !(job.subscribers) ~f:(fun subscriber -> + enqueue_progress subscriber progress ~terminal ) + in + job.subscribers := if terminal then [] else subscribers + +let remove_subscriber job subscriber = + job.subscribers := + List.filter !(job.subscribers) ~f:(fun subscriber' -> + not (phys_equal subscriber subscriber') ) + +let update_job job ~status ?error ?started_at ?finished_at ?diffs_published () = + job.status <- status ; + Option.iter error ~f:(fun value -> job.error <- Some value) ; + Option.iter started_at ~f:(fun value -> job.started_at <- Some value) ; + Option.iter finished_at ~f:(fun value -> job.finished_at <- Some value) ; + Option.iter diffs_published ~f:(fun value -> job.diffs_published <- value) ; + notify_subscribers job + +let backfill_kind ~from_hash ~index = + if is_genesis_hash from_hash && Int.equal index 0 + then Explorer_events.Transaction_kind.Genesis_replay + else Explorer_events.Transaction_kind.Sync_replay + +let source_query from_hash = + if is_genesis_hash from_hash then `Genesis + else `Specific from_hash + +let create ~logger ~da_config ~nats_url = + let%map nats_client = + Explorer_events.connect_and_ensure_jetstream_strict ~logger nats_url + in + let nats_client = + match nats_client with + | Ok client -> + [%log debug] "Explorer backfill service connected to NATS: url=%s" + (Uri.to_string nats_url) ; + client + | Error error -> + failwithf "Explorer backfill service requires JetStream: %s" + (Error.to_string_hum error) () + in + { logger + ; da_config + ; nats_client = Some nats_client + ; jobs = String.Table.create () + ; instance_id = Uuid.to_string (Uuid_unix.create ()) + ; started_at = Time.now () + } + +let shutdown t = + match t.nats_client with + | None -> + Deferred.unit + | Some client -> + let logger = t.logger in + [%log debug] "Shutting down explorer backfill NATS client: instance_id=%s" + t.instance_id ; + Nats_client_async.close client + +let find_job t id = Hashtbl.find t.jobs id + +let prune_terminal_jobs t = + let cutoff = Time.sub (Time.now ()) terminal_job_retention in + Hashtbl.filter_inplace t.jobs ~f:(fun job -> + (not (is_terminal job.status)) + || + match job.finished_at with + | None -> + true + | Some finished_at -> + Time.( > ) finished_at cutoff ) + +let find_active_job_by_range t ~from_hash ~to_hash = + Hashtbl.data t.jobs + |> List.find ~f:(fun job -> + (not (is_terminal job.status)) + && Ledger_hash.equal job.from_hash from_hash + && Ledger_hash.equal job.to_hash to_hash ) + +let active_job_count t = + Hashtbl.count t.jobs ~f:(fun job -> not (is_terminal job.status)) + +let subscribe_progress t ~id = + match find_job t id with + | None -> + Or_error.errorf "Unknown backfill job %s" id + | Some job -> + let reader, writer = Pipe.create () in + let subscriber = create_subscriber writer in + if not (is_terminal job.status) then ( + job.subscribers := subscriber :: !(job.subscribers) ; + don't_wait_for + ( Pipe.closed reader + >>| fun () -> remove_subscriber job subscriber ) ) ; + ignore + (enqueue_progress subscriber (progress_of_job job) + ~terminal:(is_terminal job.status) : bool ) ; + Ok reader + +let publish_message_result client ({ subject; headers; payload } : + Explorer_events.message ) = + let headers = + match headers with + | [] -> + None + | headers -> + Some (Nats_client.Headers.of_list headers) + in + Nats_client_async.publish_result client ~subject ?headers + (Yojson.Safe.to_string payload) + +let publish_backfill_diff t ~from_hash ~index ~target_ledger_hash diff = + let logger = t.logger in + let genesis = is_genesis_hash from_hash && Int.equal index 0 in + let kind = backfill_kind ~from_hash ~index in + [%log debug] + "Publishing explorer backfill transaction event: subject=%s kind=%s \ + from_hash=%s target_ledger_hash=%s index=%d genesis=%b" + Explorer_events.Subject.transactions + (Explorer_events.Transaction_kind.to_string kind) + (Ledger_hash.to_decimal_string from_hash) + (Ledger_hash.to_decimal_string target_ledger_hash) + index genesis ; + let message = + Explorer_events.build_transaction_message + ~kind ~target_ledger_hash ~genesis ~diff + in + match t.nats_client with + | None -> + [%log debug] + "Dropped explorer backfill transaction event: no NATS client \ + target_ledger_hash=%s" + (Ledger_hash.to_decimal_string target_ledger_hash) ; + `Dropped + | Some client -> + let result = publish_message_result client message in + [%log debug] + "Explorer backfill transaction event publish result: \ + target_ledger_hash=%s result=%s" + (Ledger_hash.to_decimal_string target_ledger_hash) + (Format.asprintf "%a" Nats_client_async.pp_publish_result result) ; + result + +let handle_backfill_publish_result t job ~target_ledger_hash result = + let logger = t.logger in + match result with + | `Queued -> + update_job job ~status:Running + ~diffs_published:(job.diffs_published + 1) + () ; + [%log debug] + "Backfill queued explorer transaction event: job_id=%s \ + target_ledger_hash=%s diffs_published=%d" + job.id + (Ledger_hash.to_decimal_string target_ledger_hash) + job.diffs_published ; + Ok target_ledger_hash + | `Dropped -> + [%log debug] + "Backfill failed to queue explorer transaction event: job_id=%s \ + target_ledger_hash=%s" + job.id + (Ledger_hash.to_decimal_string target_ledger_hash) ; + Error (Error.of_string "NATS publish dropped while backfilling diffs") + +let source_hash = function + | `Genesis -> + genesis_hash + | `Specific ledger_hash -> + ledger_hash + +let backfill_intervals t ~source_ledger_hash ~target_ledger_hash = + let rec get_intervals ~target_ledger_hash = + let%bind.Deferred.Result chain = + Da_layer.Client.get_ledger_hashes_chain ~logger:t.logger + ~config:t.da_config ~max_length:backfill_interval_size + ~source_ledger_hash:(`Specific source_ledger_hash) ~target_ledger_hash + () + in + match chain with + | [] -> + return (Ok []) + | [ last ] -> + return (Ok [ (source_ledger_hash, last) ]) + | chain -> + let interval_source = List.hd_exn chain in + let interval_target = List.last_exn chain in + let%bind.Deferred.Result intervals = + get_intervals ~target_ledger_hash:interval_source + in + return (Ok ((interval_source, interval_target) :: intervals)) + in + get_intervals ~target_ledger_hash >>| Result.map ~f:List.rev + +let publish_target t job ~current_source ~index ~target_ledger_hash = + let logger = t.logger in + [%log debug] + "Backfill fetching diff before explorer publish: job_id=%s \ + current_source=%s target_ledger_hash=%s index=%d" + job.id + (Ledger_hash.to_decimal_string current_source) + (Ledger_hash.to_decimal_string target_ledger_hash) + index ; + let%bind.Deferred.Result diff = + Da_layer.Client.get_diff ~logger:t.logger ~config:t.da_config + ~ledger_hash:target_ledger_hash + in + if + not + (Ledger_hash.equal + (Da_layer.Diff.Stable.V4.source_ledger_hash diff) + current_source ) + then + Deferred.return + (Error + (Error.of_string + "Backfill diff chain does not match requested ledger hash \ + progression") ) + else + publish_backfill_diff t ~from_hash:job.from_hash ~index ~target_ledger_hash + diff + |> handle_backfill_publish_result t job ~target_ledger_hash + |> Deferred.return + +let run_job t job = + let logger = t.logger in + let now = Time.now () in + [%log debug] + "Starting explorer backfill job: job_id=%s from_hash=%s to_hash=%s" + job.id + (Ledger_hash.to_decimal_string job.from_hash) + (Ledger_hash.to_decimal_string job.to_hash) ; + update_job job ~status:Running ~started_at:now () ; + let source = source_query job.from_hash in + don't_wait_for + (Monitor.try_with_or_error (fun () -> + let source_ledger_hash = source_hash source in + let%bind.Deferred.Result intervals = + backfill_intervals t ~source_ledger_hash + ~target_ledger_hash:job.to_hash + in + let index = ref 0 in + let%bind.Deferred.Result final_source = + Deferred.List.fold intervals ~init:(Ok source_ledger_hash) + ~f:(fun acc (interval_source, interval_target) -> + match acc with + | Error _ as error -> + return error + | Ok current_source -> + if not (Ledger_hash.equal current_source interval_source) + then + return + (Error + (Error.of_string + "Backfill intervals do not match requested ledger \ + hash progression") ) + else + let%bind.Deferred.Result target_ledger_hashes = + Da_layer.Client.get_ledger_hashes_chain ~logger:t.logger + ~config:t.da_config ~max_length:backfill_interval_size + ~source_ledger_hash:(`Specific interval_source) + ~target_ledger_hash:interval_target () + in + Deferred.List.fold target_ledger_hashes + ~init:(Ok interval_source) + ~f:(fun acc target_ledger_hash -> + match acc with + | Error _ as error -> + return error + | Ok current_source -> + let%map result = + publish_target t job ~current_source ~index:!index + ~target_ledger_hash + in + (match result with Ok _ -> incr index | Error _ -> ()) ; + result ) ) + in + if Ledger_hash.equal final_source job.to_hash + then Deferred.Result.return () + else + Deferred.return + (Error + (Error.of_string + "Ledger hash chain ended before reaching backfill target") ) ) + >>= function + | Ok (Ok ()) -> + [%log debug] + "Completed explorer backfill job: job_id=%s diffs_published=%d" + job.id job.diffs_published ; + update_job job ~status:Completed ~finished_at:(Time.now ()) () ; + Deferred.unit + | Ok (Error error) | Error error -> + [%log debug] + "Failed explorer backfill job: job_id=%s error=%s" + job.id (Error.to_string_hum error) ; + update_job job ~status:Failed ~finished_at:(Time.now ()) + ~error:(Error.to_string_hum error) () ; + Deferred.unit ) + +let parse_ledger_hash hash = + Or_error.try_with (fun () -> Ledger_hash.of_decimal_string hash) + |> Result.map_error ~f:(fun err -> + Error.tag_arg err "Invalid ledger hash" hash String.sexp_of_t ) + +let create_job ?error ?started_at ?finished_at ~status ~from_hash ~to_hash () = + { id = Uuid.to_string (Uuid_unix.create ()) + ; from_hash + ; to_hash + ; status + ; diffs_published = 0 + ; error + ; created_at = Time.now () + ; started_at + ; finished_at + ; subscribers = ref [] + } + +let register_job t job = + let logger = t.logger in + Hashtbl.set t.jobs ~key:job.id ~data:job ; + [%log debug] + "Registered explorer backfill job: job_id=%s from_hash=%s to_hash=%s" + job.id + (Ledger_hash.to_decimal_string job.from_hash) + (Ledger_hash.to_decimal_string job.to_hash) ; + notify_subscribers job ; + job + +let start_backfill t ~from_hash ~to_hash = + let logger = t.logger in + prune_terminal_jobs t ; + match find_active_job_by_range t ~from_hash ~to_hash with + | Some job -> + [%log debug] + "Reusing active explorer backfill job: job_id=%s from_hash=%s to_hash=%s" + job.id + (Ledger_hash.to_decimal_string from_hash) + (Ledger_hash.to_decimal_string to_hash) ; + job + | None -> + if active_job_count t >= max_active_backfill_jobs then ( + let now = Time.now () in + [%log debug] + "Rejecting explorer backfill job because active job limit is reached: \ + from_hash=%s to_hash=%s max_active=%d" + (Ledger_hash.to_decimal_string from_hash) + (Ledger_hash.to_decimal_string to_hash) + max_active_backfill_jobs ; + create_job ~status:Failed ~from_hash ~to_hash + ~error:"Another backfill job is already active" ~started_at:now + ~finished_at:now () + |> register_job t ) + else + let job = create_job ~status:Queued ~from_hash ~to_hash () in + let job = register_job t job in + run_job t job ; + job + +let failed_job_snapshot_from_strings ~from_hash ~to_hash error = + let now = Time.now () in + { id = Uuid.to_string (Uuid_unix.create ()) + ; from_hash + ; to_hash + ; status = string_of_job_status Failed + ; diffs_published = 0 + ; error = Some error + ; created_at = timestamp_string now + ; started_at = None + ; finished_at = Some (timestamp_string now) + } + +let start_backfill_from_strings t ~from_hash ~to_hash = + let open Or_error.Let_syntax in + let%bind from_hash = parse_ledger_hash from_hash in + let%map to_hash = parse_ledger_hash to_hash in + start_backfill t ~from_hash ~to_hash diff --git a/src/app/zeko/sequencer/explorer/explorer_backfill_sse.ml b/src/app/zeko/sequencer/explorer/explorer_backfill_sse.ml new file mode 100644 index 0000000000..4cce257e3f --- /dev/null +++ b/src/app/zeko/sequencer/explorer/explorer_backfill_sse.ml @@ -0,0 +1,78 @@ +(* Implements the GraphQL-SSE transport for the standalone backfill server. *) + +open Core +open Async + +let headers = + Cohttp.Header.of_list + [ ("Content-Type", "text/event-stream") + ; ("Cache-Control", "no-cache") + ; ("Connection", "keep-alive") + ] + +let next_event payload = + "event: next\ndata: " ^ Yojson.Basic.to_string payload ^ "\n\n" + +let complete_event = "event: complete\ndata: {}\n\n" + +let parse_request req body = + Init.Graphql_internal.Params.extract req body + |> Result.map_error ~f:Error.of_string + +let execute_subscription t req body = + let open Deferred.Let_syntax in + match parse_request req body with + | Error err -> + Deferred.return (Error err) + | Ok (query, variables, operation_name) -> ( + match Graphql_parser.parse query with + | Error err -> + Deferred.return (Error (Error.of_string err)) + | Ok doc -> + let%map result = + Graphql_async.Schema.execute Explorer_backfill_graphql.schema t + ?variables ?operation_name doc + in + match result with + | Ok (`Stream stream) -> + Ok stream + | Ok (`Response _) -> + Error + (Error.of_string + "Expected a GraphQL subscription for /graphql/stream") + | Error err -> + Error + (Error.of_string + ("Invalid GraphQL subscription: " + ^ Yojson.Basic.to_string err ) ) ) + +let rec write_stream body_writer stream = + let open Deferred.Let_syntax in + match%bind Pipe.read stream with + | `Eof -> + let%map () = Pipe.write body_writer complete_event in + Pipe.close body_writer + | `Ok payload -> + let payload = + match payload with + | Ok payload -> + payload + | Error err -> + err + in + let%bind () = Pipe.write body_writer (next_event payload) in + write_stream body_writer stream + +let callback t _conn req body = + let open Deferred.Let_syntax in + let%bind body = Cohttp_async.Body.to_string body in + match%bind execute_subscription t req body with + | Error err -> + Cohttp_async.Server.respond_string ~status:`Bad_request + (Error.to_string_hum err) + >>| fun response -> `Response response + | Ok stream -> + let body_reader, body_writer = Pipe.create () in + don't_wait_for (write_stream body_writer stream) ; + Cohttp_async.Server.respond_with_pipe ~headers body_reader + >>| fun response -> `Response response diff --git a/src/app/zeko/sequencer/explorer/explorer_events.ml b/src/app/zeko/sequencer/explorer/explorer_events.ml new file mode 100644 index 0000000000..ccf771a102 --- /dev/null +++ b/src/app/zeko/sequencer/explorer/explorer_events.ml @@ -0,0 +1,479 @@ +(* Defines the shared explorer-facing NATS subjects, payload encoding, and + publishing helpers used by both the live sequencer path and backfill jobs. *) + +open Core_kernel +open Async +open Mina_base + +module Transaction_kind = struct + type t = User_command | Fee_transfer | Sync_replay | Genesis_replay + + let to_string = function + | User_command -> + "user_command" + | Fee_transfer -> + "fee_transfer" + | Sync_replay -> + "sync_replay" + | Genesis_replay -> + "genesis_replay" +end + +module Finality_status = struct + type t = Proved | Committed + + let to_string = function Proved -> "proved" | Committed -> "committed" +end + +module Subject = struct + let transactions = "zeko.l2.transactions" + + let finality = "zeko.l2.finality" + + let health = "zeko.health" +end + +module Jetstream = struct + type stream = + { name : string + ; subjects : string list + ; max_msgs : int + ; max_age_ns : int + ; duplicate_window_ns : int option + } + + let duplicate_window_ns = 120_000_000_000 + + let l2_stream = + { name = "zeko-l2" + ; subjects = [ "zeko.l2.>" ] + ; max_msgs = -1 + ; max_age_ns = 7_776_000_000_000_000 + ; duplicate_window_ns = Some duplicate_window_ns + } + + let health_stream = + { name = "zeko-health" + ; subjects = [ Subject.health ] + ; max_msgs = 1_000 + ; max_age_ns = 0 + ; duplicate_window_ns = None + } + + let streams = [ l2_stream; health_stream ] + + let stream_name = l2_stream.name + + let stream_config stream = + let duplicate_window = + match stream.duplicate_window_ns with + | None -> + [] + | Some duplicate_window_ns -> + [ ("duplicate_window", `Int duplicate_window_ns) ] + in + `Assoc + ( [ ("name", `String stream.name) + ; ( "subjects" + , `List (List.map stream.subjects ~f:(fun subject -> `String subject)) + ) + ; ("retention", `String "limits") + ; ("storage", `String "file") + ; ("discard", `String "old") + ; ("max_msgs", `Int stream.max_msgs) + ; ("max_bytes", `Int (-1)) + ; ("max_age", `Int stream.max_age_ns) + ; ("max_msgs_per_subject", `Int (-1)) + ; ("max_msg_size", `Int (-1)) + ; ("num_replicas", `Int 1) + ] + @ duplicate_window ) + + let api_subject ?(stream_name = stream_name) operation = + sprintf "$JS.API.STREAM.%s.%s" operation stream_name +end + +type message = + { subject : string + ; headers : (string * string) list + ; payload : Yojson.Safe.t + } + +type sink = message -> unit + +let noop_sink _ = () + +let timestamp_json ~logger = + Block_time.now (Block_time.Controller.basic ~logger) |> Block_time.to_yojson + +let nats_msg_id target_ledger_hash = + Ledger_hash.to_decimal_string target_ledger_hash + +let nats_msg_id_headers target_ledger_hash = + [ ("Nats-Msg-Id", nats_msg_id target_ledger_hash) ] + +let finality_nats_msg_id ~status target_ledger_hash = + sprintf "finality-%s-%s" + (Finality_status.to_string status) + (Ledger_hash.to_decimal_string target_ledger_hash) + +let finality_nats_msg_id_headers ~status target_ledger_hash = + [ ("Nats-Msg-Id", finality_nats_msg_id ~status target_ledger_hash) ] + +let create_nats_sink ?logger client : sink = + fun { subject; headers; payload } -> + let headers = Nats_client.Headers.of_list headers in + match + Nats_client_async.publish_result client ~subject ~headers + (Yojson.Safe.to_string payload) + with + | `Queued -> + () + | `Dropped -> + Option.iter logger ~f:(fun logger -> + [%log warn] "Dropped explorer NATS message" + ~metadata:[ ("subject", `String subject) ] ) + +let jetstream_request_timeout = Time_ns.Span.of_sec 5. + +let warn_jetstream ?(stream_name = Jetstream.stream_name) ~logger message + ~metadata = + [%log warn] "%s" message + ~metadata:(("stream", `String stream_name) :: metadata) + +let response_error json = + match json with + | `Assoc fields -> ( + match List.Assoc.find fields "error" ~equal:String.equal with + | Some (`Assoc error_fields) -> + Some error_fields + | _ -> + None ) + | _ -> + None + +let response_error_int fields name = + match List.Assoc.find fields name ~equal:String.equal with + | Some (`Int value) -> + Some value + | _ -> + None + +let response_error_string fields name = + match List.Assoc.find fields name ~equal:String.equal with + | Some (`String value) -> + Some value + | _ -> + None + +let is_stream_not_found fields = + Option.value_map + (response_error_int fields "err_code") + ~default:false ~f:(Int.equal 10059) + || Option.value_map + (response_error_int fields "code") + ~default:false ~f:(Int.equal 404) + +let jetstream_request client ~stream_name ~operation payload = + let subject = Jetstream.api_subject ~stream_name operation in + Nats_client_async.request client ~subject ~timeout:jetstream_request_timeout + (Yojson.Safe.to_string payload) + +let jetstream_response_result ~operation response = + match Or_error.try_with (fun () -> Yojson.Safe.from_string response) with + | Error error -> + Error (Error.tag error ~tag:"Could not parse JetStream response") + | Ok json -> ( + match response_error json with + | None -> + Ok () + | Some fields -> + let description = + Option.value + (response_error_string fields "description") + ~default:"unknown JetStream error" + in + Or_error.errorf "JetStream stream %s failed: %s" operation + description ) + +let create_or_update_jetstream_stream_result client stream ~operation = + jetstream_request client ~stream_name:stream.Jetstream.name ~operation + (Jetstream.stream_config stream) + >>| function + | Error error -> + Error + (Error.tag error + ~tag:(sprintf "JetStream stream %s request failed" operation) ) + | Ok response -> + jetstream_response_result ~operation response + +let ensure_jetstream_stream_result_one client stream = + let open Deferred.Let_syntax in + let%bind info = + jetstream_request client ~stream_name:stream.Jetstream.name + ~operation:"INFO" (`Assoc []) + in + match info with + | Error error -> + return (Error (Error.tag error ~tag:"JetStream stream INFO request failed")) + | Ok response -> ( + match Or_error.try_with (fun () -> Yojson.Safe.from_string response) with + | Error error -> + return + (Error + (Error.tag error + ~tag:"Could not parse JetStream stream INFO response" ) ) + | Ok json -> ( + match response_error json with + | Some fields when is_stream_not_found fields -> + create_or_update_jetstream_stream_result client stream + ~operation:"CREATE" + | Some fields -> + let description = + Option.value + (response_error_string fields "description") + ~default:"unknown JetStream error" + in + Deferred.return + (Or_error.errorf "JetStream stream INFO failed: %s" + description ) + | None -> + create_or_update_jetstream_stream_result client stream + ~operation:"UPDATE" ) ) + +let ensure_jetstream_stream_result client = + let rec go = function + | [] -> + Deferred.return (Ok ()) + | stream :: streams -> ( + let%bind result = ensure_jetstream_stream_result_one client stream in + match result with + | Error _ as error -> + Deferred.return error + | Ok () -> + go streams ) + in + go Jetstream.streams + +let ensure_jetstream_stream ~logger client = + ensure_jetstream_stream_result client >>| function + | Ok () -> + () + | Error error -> + warn_jetstream ~logger "JetStream stream setup failed" + ~metadata:[ ("error", `String (Error.to_string_hum error)) ] + +let nats_uri_available ?(timeout = Time.Span.of_sec 3.) ~logger uri = + match Uri.host uri with + | None -> + warn_jetstream ~logger "NATS URL is missing a host" + ~metadata:[ ("uri", `String (Uri.to_string uri)) ] ; + Deferred.return false + | Some host -> ( + let port = Uri.port uri |> Option.value ~default:4222 in + let where = + Tcp.Where_to_connect.of_host_and_port (Host_and_port.create ~host ~port) + in + Clock.with_timeout timeout + (Monitor.try_with_or_error (fun () -> + Tcp.with_connection where ~timeout (fun _ _ _ -> Deferred.unit) ) ) + >>| function + | `Timeout -> + warn_jetstream ~logger + "Timed out checking NATS availability; explorer publishing disabled" + ~metadata:[ ("uri", `String (Uri.to_string uri)) ] ; + false + | `Result (Error error) -> + warn_jetstream ~logger + "NATS is unavailable; explorer publishing disabled" + ~metadata: + [ ("uri", `String (Uri.to_string uri)) + ; ("error", `String (Error.to_string_hum error)) + ] ; + false + | `Result (Ok ()) -> + true ) + +let connect_and_ensure_jetstream ?(timeout = Time_ns.Span.of_sec 5.) ~logger uri + = + let open Deferred.Let_syntax in + let%bind available = nats_uri_available ~logger uri in + if not available then return None + else + let connect = + Monitor.try_with_or_error (fun () -> Nats_client_async.connect (Some uri)) + in + let%bind connected = + Clock_ns.with_timeout timeout connect + in + match connected with + | `Timeout -> + Deferred.upon connect (function + | Ok client -> + don't_wait_for (Nats_client_async.close client) + | Error _ -> + () ) ; + warn_jetstream ~logger + "Timed out connecting to NATS; explorer publishing disabled" + ~metadata:[ ("uri", `String (Uri.to_string uri)) ] ; + return None + | `Result (Error error) -> + warn_jetstream ~logger + "Failed to connect to NATS; explorer publishing disabled" + ~metadata: + [ ("uri", `String (Uri.to_string uri)) + ; ("error", `String (Error.to_string_hum error)) + ] ; + return None + | `Result (Ok client) -> + let%map ensure_result = + Monitor.try_with_or_error (fun () -> + ensure_jetstream_stream ~logger client ) + in + Result.iter_error ensure_result ~f:(fun error -> + warn_jetstream ~logger "JetStream setup raised an exception" + ~metadata:[ ("error", `String (Error.to_string_hum error)) ] ) ; + Some client + +let connect_and_ensure_jetstream_strict ?timeout ~logger uri = + let open Deferred.Let_syntax in + let%bind client = connect_and_ensure_jetstream ?timeout ~logger uri in + match client with + | None -> + Deferred.return + (Or_error.errorf "Could not connect to NATS at %s" (Uri.to_string uri)) + | Some client -> + let%map result = ensure_jetstream_stream_result client in + Result.map_error result ~f:(fun error -> + don't_wait_for (Nats_client_async.close client) ; + Error.tag error ~tag:"Could not configure JetStream" ) + |> Result.map ~f:(fun () -> client) + +let assoc_field_exn fields name = + List.Assoc.find_exn fields name ~equal:String.equal + +let transaction_command_payload actions = + match actions with + | `Actions _ -> + `Null + | `Command_with_action_step_flags (command, action_step_flags) -> + let command_type, command_json = + match command with + | User_command.Signed_command command -> + ("signed_command", Signed_command.Stable.V2.to_yojson command) + | User_command.Zkapp_command command -> + ("zkapp_command", Zkapp_command.Stable.V1.to_yojson command) + in + `Assoc + [ ("type", `String command_type) + ; ("raw", command_json) + ; ( "action_step_flags" + , `List (List.map action_step_flags ~f:(fun flag -> `Bool flag)) + ) + ] + +let changed_accounts_payload changed_accounts = + `List + (List.map changed_accounts ~f:(fun (index, account) -> + `Assoc + [ ("index", `Int index) + ; ("account", Account.Stable.V2.to_yojson account) + ] ) ) + +let legacy_command_with_action_step_flags_payload actions = + match actions with + | `Actions _ -> + `Null + | `Command_with_action_step_flags (command, action_step_flags) -> + `List + [ User_command.Stable.V2.to_yojson command + ; `List (List.map action_step_flags ~f:(fun flag -> `Bool flag)) + ] + +let build_transaction_message ~kind ~target_ledger_hash ~genesis ~diff = + let diff_json = Da_layer.Diff.Stable.V4.to_yojson diff in + let diff_fields = + match diff_json with + | `Assoc fields -> + fields + | _ -> + [] + in + let actions = Da_layer.Diff.Stable.V4.actions diff in + { subject = Subject.transactions + ; headers = nats_msg_id_headers target_ledger_hash + ; payload = + `Assoc + [ ("kind", `String (Transaction_kind.to_string kind)) + ; ("source_ledger_hash", assoc_field_exn diff_fields "source_ledger_hash") + ; ("target_ledger_hash", Ledger_hash.to_yojson target_ledger_hash) + ; ("timestamp", assoc_field_exn diff_fields "timestamp") + ; ("acc_set", assoc_field_exn diff_fields "acc_set") + ; ("command", transaction_command_payload actions) + ; ( "changed_accounts" + , changed_accounts_payload + (Da_layer.Diff.Stable.V4.changed_accounts diff) ) + ; ( "command_with_action_step_flags" + , legacy_command_with_action_step_flags_payload actions ) + ; ("actions", assoc_field_exn diff_fields "actions") + ; ("genesis", `Bool genesis) + ; ("diff", diff_json) + ] + } + +let build_live_diff ~logger ~diff ~acc_set_root = + Da_layer.Diff.add_time_and_acc_set ~logger diff ~acc_set:acc_set_root + +let build_finality_message ~logger ~status ~source_ledger_hash + ~target_ledger_hash = + { subject = Subject.finality + ; headers = finality_nats_msg_id_headers ~status target_ledger_hash + ; payload = + `Assoc + [ ("status", `String (Finality_status.to_string status)) + ; ("level", `String (Finality_status.to_string status)) + ; ("ledger_hash", Ledger_hash.to_yojson target_ledger_hash) + ; ("source_ledger_hash", Ledger_hash.to_yojson source_ledger_hash) + ; ("target_ledger_hash", Ledger_hash.to_yojson target_ledger_hash) + ; ("timestamp", timestamp_json ~logger) + ] + } + +let build_health_message ~logger ~service ~instance_id ~status + ?last_published_hash ?unproved_hash () = + let optional_hash name = function + | None -> + [] + | Some hash -> + [ (name, Ledger_hash.to_yojson hash) ] + in + { subject = Subject.health + ; headers = [] + ; payload = + `Assoc + ( [ ("service", `String service) + ; ("instance_id", `String instance_id) + ; ("status", `String status) + ] + @ optional_hash "last_published_hash" last_published_hash + @ optional_hash "unproved_hash" unproved_hash + @ [ ("timestamp", timestamp_json ~logger) ] ) + } + +let publish sink message = sink message + +let publish_transaction sink ~kind ~target_ledger_hash ~genesis ~diff = + publish sink + @@ build_transaction_message ~kind ~target_ledger_hash ~genesis ~diff + +let publish_finality sink ~logger ~status ~source_ledger_hash + ~target_ledger_hash = + publish sink + @@ build_finality_message ~logger ~status ~source_ledger_hash + ~target_ledger_hash + +let publish_health sink ~logger ~service ~instance_id ~status + ?last_published_hash ?unproved_hash () = + publish sink + @@ build_health_message ~logger ~service ~instance_id ~status + ?last_published_hash ?unproved_hash () diff --git a/src/app/zeko/sequencer/explorer/tests/README.md b/src/app/zeko/sequencer/explorer/tests/README.md new file mode 100644 index 0000000000..7744d36c60 --- /dev/null +++ b/src/app/zeko/sequencer/explorer/tests/README.md @@ -0,0 +1,114 @@ +# Explorer Gherkin Tests + +These tests cover the explorer-facing behavior added by the sequencer NATS and +backfill work. The feature files stay in Gherkin so the behavior is readable, +while the OCaml harnesses execute those scenarios against the in-repo modules +and, where needed, a real NATS server. + +## Explorer Gherkin Integration Suite + +There is one explorer Gherkin integration suite with two execution categories +for now. + +### In-process scenarios + +File: + +- `explorer_gherkin_tests.ml` + +This is an inline-test library. It checks behavior that does not require an +external broker or service process: + +- explorer event payload and header contracts +- backfill GraphQL schema behavior +- GraphQL-SSE progress stream formatting +- sequencer replay genesis classification + +Run it directly with: + +```bash +opam exec -- env -u DUNE_RPC dune runtest --profile=devnet src/app/zeko/sequencer/explorer/tests +``` + +This command also runs the broker-backed scenarios when `NATS_URL` points at a +running NATS server, because the tests directory wires both current entry points +into the Dune `runtest` alias. + +### Broker-backed scenarios + +File: + +- `explorer_nats_gherkin_tests.ml` + +This is a standalone Async executable because it connects to a real NATS server. +It checks that the shared explorer publisher paths produce messages that a real +subscriber receives with the expected subject, payload, and `Nats-Msg-Id` +header. It also checks the Zeko JetStream stream contract by publishing the +same transaction message twice with the same `Nats-Msg-Id` and asserting the +stream stores one message. + +Run it separately with: + +```bash +NATS_URL="nats://127.0.0.1:4222" \ + opam exec -- env -u DUNE_RPC dune exec --profile=devnet \ + src/app/zeko/sequencer/explorer/tests/explorer_nats_gherkin_tests.exe +``` + +Run the whole explorer suite with NATS enabled: + +```bash +NATS_URL="nats://127.0.0.1:4222" \ + opam exec -- env -u DUNE_RPC dune runtest --profile=devnet \ + src/app/zeko/sequencer/explorer/tests +``` + +The `Explorer Gherkin` CI workflow uses this full-suite command and provides +NATS through a GitHub Actions service container with JetStream enabled. + +## Adding Coverage + +Keep scenarios declarative: describe the explorer behavior, not the OCaml +implementation steps. Add one scenario per behavior. + +Use in-process scenarios for deterministic behavior that does not need external +infrastructure. Use broker-backed scenarios when the behavior depends on NATS +delivery, subjects, headers, or JetStream storage semantics. + +## Process-level E2E Scenarios + +File: + +- `explorer_e2e_gherkin_tests.ml` + +Process-level end-to-end coverage is a separate, slower Gherkin entry point +rather than part of the fast integration suite. It should boot real services +and assert behavior from the outside where practical: + +- start the sequencer with `--nats-url`, submit a transaction through the + sequencer API, and assert a NATS subscriber receives `zeko.l2.transactions` +- start the backfill HTTP server with a DA fixture and real NATS, call the + GraphQL `backfill` mutation over HTTP, and assert both GraphQL-SSE progress + and replayed NATS messages + +The current E2E script starts the same service shape as the sequencer +integration test: real L1 test ledger, DA nodes, signers, RabbitMQ, Postgres, +and NATS. The sequencer runtime is still driven through `Sequencer.create`, +matching the existing sequencer integration tests. A later follow-up can move +this to a fully external `run.exe` process once the bootstrap/deploy path is +ready for that shape. + +The explorer E2E script is separate from `run-sequencer-test.sh` so the +existing sequencer harness keeps its original interface and the explorer suite +can build and run its dedicated Gherkin executable explicitly. + +Run it with: + +```bash +NATS_URL="nats://127.0.0.1:4222" \ + ./src/app/zeko/sequencer/explorer/tests/run-explorer-e2e-test.sh real 1 +``` + +The E2E harness should remain separate from the faster integration suite so +failures identify whether the break is in the in-process behavior, NATS broker +delivery, or full service orchestration. diff --git a/src/app/zeko/sequencer/explorer/tests/dune b/src/app/zeko/sequencer/explorer/tests/dune new file mode 100644 index 0000000000..ca7552d8ef --- /dev/null +++ b/src/app/zeko/sequencer/explorer/tests/dune @@ -0,0 +1,92 @@ +; Gherkin-backed acceptance tests for the explorer publish and backfill contract. + +(library + (name explorer_feature_parser) + (wrapped false) + (modules feature_parser) + (libraries core)) + +(library + (name explorer_gherkin_tests) + (inline_tests) + (modules explorer_gherkin_tests) + (libraries + explorer_feature_parser + explorer_events + explorer_backfill + sequencer_lib + is_compile_simple_real_real + da_layer + zeko_constants + ;; mina ;; + mina_base + snark_params + ;; opam libraries ;; + graphql-async + graphql_parser + yojson + core + core_kernel + async + async_unix + nats-client + nats-client-async + ppx_inline_test.config) + (preprocess + (pps ppx_let ppx_jane ppx_mina))) + +(executable + (name explorer_nats_gherkin_tests) + (modules explorer_nats_gherkin_tests) + (libraries + explorer_feature_parser + explorer_events + explorer_backfill + sequencer_lib + is_compile_simple_real_real + da_layer + zeko_constants + ;; mina ;; + mina_base + snark_params + ;; opam libraries ;; + yojson + core + core_kernel + async + async_unix + nats-client + nats-client-async) + (preprocess + (pps ppx_let ppx_jane ppx_mina))) + +(executable + (name explorer_e2e_gherkin_tests) + (modules explorer_e2e_gherkin_tests) + (libraries + explorer_feature_parser + explorer_events + explorer_backfill + sequencer_lib + test_spec + is_compile_simple_real_real + da_layer + zeko_constants + ;; mina ;; + mina_base + snark_params + ;; opam libraries ;; + yojson + core + core_kernel + async + async_unix + nats-client + nats-client-async) + (preprocess + (pps ppx_let ppx_jane ppx_mina))) + +(rule + (alias runtest) + (action + (run ./explorer_nats_gherkin_tests.exe))) diff --git a/src/app/zeko/sequencer/explorer/tests/explorer_e2e_gherkin_tests.ml b/src/app/zeko/sequencer/explorer/tests/explorer_e2e_gherkin_tests.ml new file mode 100644 index 0000000000..39f60f0e20 --- /dev/null +++ b/src/app/zeko/sequencer/explorer/tests/explorer_e2e_gherkin_tests.ml @@ -0,0 +1,281 @@ +(* Runs slow Gherkin-backed explorer E2E scenarios against the same external + services used by the sequencer integration test: L1 test ledger, DA nodes, + signers, RabbitMQ, Postgres, and a real NATS broker. *) + +open Core +open Async +open Mina_base +open Sequencer_lib +open Zeko_sequencer +open Test_spec +open Handle.Operator + +let logger = + Cli_lib.Stdout_log.setup false Logger.Level.Spam ; + Logger.create () + +let run = Thread_safe.block_on_async_exn + +let gql_uri = Uri.of_string "http://localhost:8080/graphql" + +let nats_url () = + match Sys.getenv "NATS_URL" with + | Some value -> + Uri.of_string value + | None -> + failwith "NATS_URL must be set for explorer E2E Gherkin tests" + +let da_config = + Da_layer.Client.Config.of_string_list [ "127.0.0.1:8555"; "127.0.0.1:8556" ] + +let da_keys = + run (fun () -> Da_layer.Client.Config.fetch_public_keys ~logger da_config) + +let da_quorum = 2 + +let mq_host = Host_and_port.of_string "localhost:5672" + +let slot_acceptance = Time.Span.of_min 60. + +let postgres_uri () = + run (fun () -> + Relational_db.For_tests.create_database ~port:5433 + "explorer_e2e_gherkin" ) + +let drop_postgres () = + run (fun () -> + Relational_db.For_tests.drop_database ~port:5433 + "explorer_e2e_gherkin" ) + +let shutdown_sequencer sequencer = + Gc.full_major () ; + run (fun () -> Sequencer.shutdown !sequencer) ; + Handle.invalidate sequencer + +let connect_or_fail uri = + Clock_ns.with_timeout (Time_ns.Span.of_sec 5.) + (Nats_client_async.connect (Some uri)) + >>= function + | `Timeout -> + failwithf "timed out connecting to %s" (Uri.to_string uri) () + | `Result client -> + return client + +let expect_ok label = function + | Ok value -> + value + | Error err -> + failwithf "%s failed: %s" label (Error.to_string_hum err) () + +let subscribe_transactions subscriber = + let%map subscription = + Nats_client_async.subscribe subscriber + ~subject:Explorer_events.Subject.transactions () + in + expect_ok "transaction subscription" subscription + +let wait_for_subscription_registration () = + Clock_ns.after (Time_ns.Span.of_ms 100.) + +let read_message label reader = + Clock_ns.with_timeout (Time_ns.Span.of_sec 10.) (Pipe.read reader) + >>= function + | `Timeout -> + failwithf "%s timed out waiting for a message" label () + | `Result `Eof -> + failwithf "%s closed before delivering a message" label () + | `Result (`Ok message) -> + return message + +let read_until label reader ~f = + let rec go remaining = + if remaining <= 0 + then failwithf "%s did not observe the expected message" label () ; + let%bind message = read_message label reader in + match f message with + | Some value -> + return value + | None -> + go (remaining - 1) + in + go 20 + +let headers (message : Nats_client.Protocol.message) = + message.headers + |> Option.value ~default:Nats_client.Headers.empty + |> Nats_client.Headers.to_list + +let require_header (message : Nats_client.Protocol.message) ~name ~expected = + match List.Assoc.find (headers message) ~equal:String.equal name with + | Some value when String.equal value expected -> + () + | Some value -> + failwithf "expected header %s=%s but received %s" name expected value () + | None -> + failwithf "missing header %s" name () + +let json_assoc key json = + match json with + | `Assoc fields -> + List.Assoc.find fields key ~equal:String.equal + | _ -> + None + +let json_assoc_exn key json = + Option.value_exn (json_assoc key json) + ~message:(sprintf "missing JSON field %s" key) + +let message_payload (message : Nats_client.Protocol.message) = + Yojson.Safe.from_string message.payload + +let assert_safe_json_equal actual expected = + if not (Yojson.Safe.equal actual expected) + then + failwithf "Expected %s but got %s" (Yojson.Safe.to_string expected) + (Yojson.Safe.to_string actual) () + +let target_hash_json target_ledger_hash = + Ledger_hash.to_yojson target_ledger_hash + +let target_hash_string target_ledger_hash = + Ledger_hash.to_decimal_string target_ledger_hash + +let matching_transaction_payload + (message : Nats_client.Protocol.message) ~kind ~target_ledger_hash = + if not (String.equal message.subject Explorer_events.Subject.transactions) + then None + else + let payload = message_payload message in + match + ( json_assoc "kind" payload + , json_assoc "target_ledger_hash" payload ) + with + | Some (`String actual_kind), Some target + when String.equal actual_kind kind + && Yojson.Safe.equal target (target_hash_json target_ledger_hash) -> + Some payload + | _ -> + None + +let create_sequencer_spec ~postgres_uri = + Quickcheck.random_value + (Sequencer_spec.gen ~logger ~number_of_transactions:1 ~postgres_uri + ~gql_uri ~da_config ~da_keys ~da_quorum ~mq_host ~slot_acceptance + ~nats_url:(nats_url ()) () ) + +let apply_first_transaction sequencer specs = + let spec = List.hd_exn specs in + let command = + User_command.Signed_command + (command_send ~chain:Zeko_circuits_config.Inputs.chain_l2 spec) + in + run (fun () -> + Sequencer.apply_user_command !sequencer command >>| Or_error.ok_exn) ; + Sequencer.get_root !sequencer + +let assert_user_command_message + (message : Nats_client.Protocol.message) ~target_ledger_hash = + if + not + (String.equal message.subject Explorer_events.Subject.transactions) + then + failwithf "unexpected subject %s" message.subject () ; + require_header message ~name:"Nats-Msg-Id" + ~expected:(target_hash_string target_ledger_hash) ; + let payload = message_payload message in + assert_safe_json_equal (json_assoc_exn "kind" payload) + (`String "user_command") ; + assert_safe_json_equal (json_assoc_exn "target_ledger_hash" payload) + (target_hash_json target_ledger_hash) + +let wait_for_backfill_completion (job : Explorer_backfill_service.job) = + let rec go remaining = + if remaining <= 0 then failwith "backfill job did not complete" ; + if Explorer_backfill_service.is_terminal job.status + then return () + else Clock_ns.after (Time_ns.Span.of_ms 200.) >>= fun () -> go (remaining - 1) + in + go 100 + +let run_scenarios () = + Feature_parser.assert_scenario "e2e.feature" + "Sequencer-applied transactions are published to NATS" ; + Feature_parser.assert_scenario "e2e.feature" + "Backfill replays sequencer DA diffs to NATS" ; + let postgres_uri = postgres_uri () in + let nats_uri = nats_url () in + let subscriber = ref None in + let sequencer = ref None in + let backfill_service = ref None in + Exn.protect + ~finally:(fun () -> + Option.iter backfill_service.contents ~f:(fun service -> + run (fun () -> Explorer_backfill_service.shutdown service)) ; + Option.iter sequencer.contents ~f:(fun sequencer -> + ignore (shutdown_sequencer sequencer : (_, Handle.invalid) Handle.t)) ; + Option.iter subscriber.contents ~f:(fun client -> + run (fun () -> Nats_client_async.close client)) ; + drop_postgres ()) + ~f:(fun () -> + let client = run (fun () -> connect_or_fail nats_uri) in + subscriber := Some client ; + let live_subscription = + run (fun () -> + let%bind subscription = subscribe_transactions client in + let%map () = wait_for_subscription_registration () in + subscription) + in + let { Sequencer_spec.sequencer = sequencer_handle; specs; _ } = + create_sequencer_spec ~postgres_uri + in + sequencer := Some sequencer_handle ; + let target_ledger_hash = apply_first_transaction sequencer_handle specs in + let live_message = + run (fun () -> + read_until "sequencer publish" live_subscription.messages + ~f:(fun message -> + Option.map + (matching_transaction_payload message ~kind:"user_command" + ~target_ledger_hash ) + ~f:(fun _payload -> message))) + in + assert_user_command_message live_message ~target_ledger_hash ; + let backfill_subscription = + run (fun () -> + let%bind subscription = subscribe_transactions client in + let%map () = wait_for_subscription_registration () in + subscription) + in + let service = + run (fun () -> + Explorer_backfill_service.create ~logger ~da_config + ~nats_url:nats_uri) + in + backfill_service := Some service ; + let job = + Explorer_backfill_service.start_backfill service + ~from_hash:Explorer_backfill_service.genesis_hash + ~to_hash:target_ledger_hash + in + run (fun () -> wait_for_backfill_completion job) ; + let replay_payload = + run (fun () -> + read_until "backfill replay" backfill_subscription.messages + ~f:(fun message -> + match + matching_transaction_payload message ~kind:"sync_replay" + ~target_ledger_hash + with + | Some payload -> + require_header message ~name:"Nats-Msg-Id" + ~expected:(target_hash_string target_ledger_hash) ; + Some payload + | None -> + None)) + in + assert_safe_json_equal (json_assoc_exn "kind" replay_payload) + (`String "sync_replay") ; + printf "explorer E2E Gherkin scenarios passed against %s\n" + (Uri.to_string nats_uri)) + +let () = run_scenarios () diff --git a/src/app/zeko/sequencer/explorer/tests/explorer_gherkin_tests.ml b/src/app/zeko/sequencer/explorer/tests/explorer_gherkin_tests.ml new file mode 100644 index 0000000000..fdf0bba86a --- /dev/null +++ b/src/app/zeko/sequencer/explorer/tests/explorer_gherkin_tests.ml @@ -0,0 +1,411 @@ +(* Executes minimal Gherkin acceptance tests for the explorer-facing behavior + introduced in this PR: event payloads and backfill GraphQL/SSE behavior. *) + +open Core +open Async +open Mina_base + +let find_assoc_exn key json = + match json with + | `Assoc fields -> + List.Assoc.find_exn fields key ~equal:String.equal + | _ -> + failwith "Expected JSON object" + +let find_header_exn headers key = + List.Assoc.find_exn headers key ~equal:String.equal + +let test_service ?(instance_id = "instance-1") () : Explorer_backfill_service.t + = + { logger = Logger.create () + ; da_config = Da_layer.Client.Config.of_string_list [] + ; nats_client = None + ; jobs = String.Table.create () + ; instance_id + ; started_at = Time.epoch + } + +let graphql_response service query = + Thread_safe.block_on_async_exn (fun () -> + match Graphql_parser.parse query with + | Error err -> + failwith err + | Ok doc -> ( + Graphql_async.Schema.execute Explorer_backfill_graphql.schema service + doc + >>| function + | Ok (`Response response) -> + response + | Ok (`Stream _) -> + failwith "Expected a GraphQL response" + | Error err -> + failwith (Yojson.Basic.to_string err) ) ) + +let graphql_data_exn service query field = + find_assoc_exn field (find_assoc_exn "data" (graphql_response service query)) + +let pipe_read_exn reader = + Thread_safe.block_on_async_exn (fun () -> + Pipe.read reader + >>| function `Ok value -> value | `Eof -> failwith "Unexpected EOF" ) + +let pipe_read_string_exn reader = + Thread_safe.block_on_async_exn (fun () -> + Pipe.read reader + >>| function `Ok value -> value | `Eof -> failwith "Unexpected EOF" ) + +let add_job (service : Explorer_backfill_service.t) + ?(status = Explorer_backfill_service.Queued) ?(diffs_published = 0) ?error + ?started_at ?finished_at ?(id = "job-1") () = + let job : Explorer_backfill_service.job = + { id + ; from_hash = Explorer_backfill_service.genesis_hash + ; to_hash = Ledger_hash.empty_hash + ; status + ; diffs_published + ; error + ; created_at = Time.epoch + ; started_at + ; finished_at + ; subscribers = ref [] + } + in + Hashtbl.set service.jobs ~key:job.id ~data:job ; + job + +let sample_diff ?(source_ledger_hash = Ledger_hash.empty_hash) () = + Explorer_events.build_live_diff ~logger:(Logger.create ()) + ~diff: + (Da_layer.Diff.create_pending ~source_ledger_hash ~changed_accounts:[] + ~actions:(`Actions []) ) + ~acc_set_root:Snark_params.Tick.Field.zero + +let sse_request body = + let headers = + Cohttp.Header.of_list [ ("Content-Type", "application/json") ] + in + ( Cohttp.Request.make ~meth:`POST ~headers + (Uri.of_string "http://localhost/graphql/stream") + , body ) + +let sse_reader_exn service body = + let req, body = sse_request body in + Thread_safe.block_on_async_exn (fun () -> + match%bind + Explorer_backfill_sse.execute_subscription service req body + with + | Error err -> + failwith (Error.to_string_hum err) + | Ok stream -> + let reader, writer = Pipe.create () in + don't_wait_for (Explorer_backfill_sse.write_stream writer stream) ; + return reader ) + +let parse_sse_next_event_exn event = + let prefix = "event: next\ndata: " in + if not (String.is_prefix event ~prefix) then + failwithf "Unexpected SSE event: %s" event () + else + String.drop_prefix event (String.length prefix) + |> String.chop_suffix_exn ~suffix:"\n\n" + |> Yojson.Basic.from_string + +let backfill_progress_event_exn event = + find_assoc_exn "backfillProgress" + (find_assoc_exn "data" (parse_sse_next_event_exn event)) + +let assert_basic_json_equal actual expected = + if not (Yojson.Basic.equal actual expected) then + failwithf "Expected %s but got %s" + (Yojson.Basic.to_string expected) + (Yojson.Basic.to_string actual) + () + +let assert_safe_json_equal actual expected = + if not (Yojson.Safe.equal actual expected) then + failwithf "Expected %s but got %s" + (Yojson.Safe.to_string expected) + (Yojson.Safe.to_string actual) + () + +let%test_unit "A genesis backfill marks only the first replayed diff as genesis" + = + Feature_parser.assert_scenario "backfill-api.feature" + "A genesis backfill marks only the first replayed diff as genesis" ; + if + not + (Poly.equal + (Explorer_backfill_service.backfill_kind + ~from_hash:Explorer_backfill_service.genesis_hash ~index:0 ) + Explorer_events.Transaction_kind.Genesis_replay ) + then failwith "expected the first genesis replay diff to be marked as genesis" ; + if + not + (Poly.equal + (Explorer_backfill_service.backfill_kind + ~from_hash:Explorer_backfill_service.genesis_hash ~index:1 ) + Explorer_events.Transaction_kind.Sync_replay ) + then failwith "expected later genesis replay diffs to be marked as sync" + +let%test_unit + "The backfill mutation returns a failed job snapshot for invalid hashes" = + Feature_parser.assert_scenario "backfill-api.feature" + "The backfill mutation returns a failed job snapshot for invalid hashes" ; + let service = test_service () in + let backfill_job = + graphql_data_exn service + {|mutation { backfill(fromHash: "nonexistent", toHash: "D") { status error } }|} + "backfill" + in + assert_basic_json_equal + (find_assoc_exn "status" backfill_job) + (`String "failed") ; + [%test_eq: bool] + ( match find_assoc_exn "error" backfill_job with + | `String error -> + String.is_substring error ~substring:"Invalid ledger hash" + | _ -> + false ) + true + +let%test_unit "Backfill progress subscriptions stream job updates" = + Feature_parser.assert_scenario "backfill-api.feature" + "Backfill progress subscriptions stream job updates" ; + let service = test_service () in + let job = add_job service () in + let reader = + sse_reader_exn service + (Yojson.Basic.to_string + (`Assoc + [ ( "query" + , `String + (sprintf + {|subscription { backfillProgress(id: "%s") { id status diffsPublished error } }|} + job.id ) ) + ] ) ) + in + let initial = pipe_read_string_exn reader |> backfill_progress_event_exn in + assert_basic_json_equal (find_assoc_exn "diffsPublished" initial) (`Int 0) ; + Explorer_backfill_service.update_job job ~status:Running ~diffs_published:1 + ~started_at:Time.epoch () ; + let first = pipe_read_string_exn reader |> backfill_progress_event_exn in + assert_basic_json_equal (find_assoc_exn "diffsPublished" first) (`Int 1) ; + Explorer_backfill_service.update_job job ~status:Completed ~diffs_published:2 + ~finished_at:Time.epoch () ; + let final_progress = + pipe_read_string_exn reader |> backfill_progress_event_exn + in + assert_basic_json_equal + (find_assoc_exn "diffsPublished" final_progress) + (`Int 2) ; + assert_basic_json_equal + (find_assoc_exn "status" final_progress) + (`String "completed") + +let%test_unit "The backfill health query exposes the service instance" = + Feature_parser.assert_scenario "backfill-api.feature" + "The backfill health query exposes the service instance" ; + let health = + graphql_data_exn (test_service ()) + {|query { health { ok instanceId startedAt } }|} "health" + in + assert_basic_json_equal (find_assoc_exn "ok" health) (`Bool true) ; + assert_basic_json_equal + (find_assoc_exn "instanceId" health) + (`String "instance-1") ; + [%test_eq: bool] + ( match find_assoc_exn "startedAt" health with + | `String _ -> + true + | _ -> + false ) + true + +let%test_unit "The backfill job query returns the current job snapshot" = + Feature_parser.assert_scenario "backfill-api.feature" + "The backfill job query returns the current job snapshot" ; + let service = test_service () in + let job = add_job service ~status:Explorer_backfill_service.Running () in + let backfill_job = + graphql_data_exn service + (sprintf {|query { backfillJob(id: "%s") { id status diffsPublished } }|} + job.id ) + "backfillJob" + in + assert_basic_json_equal (find_assoc_exn "id" backfill_job) (`String job.id) ; + assert_basic_json_equal + (find_assoc_exn "status" backfill_job) + (`String "running") ; + assert_basic_json_equal (find_assoc_exn "diffsPublished" backfill_job) (`Int 0) + +let%test_unit "Concurrent backfill requests are bounded" = + Feature_parser.assert_scenario "backfill-api.feature" + "Concurrent backfill requests are bounded" ; + let service = test_service () in + let _active_job = + add_job service ~status:Explorer_backfill_service.Running () + in + let rejected_job = + Explorer_backfill_service.start_backfill service + ~from_hash:Ledger_hash.empty_hash + ~to_hash:Explorer_backfill_service.genesis_hash + in + [%test_eq: string] + (Explorer_backfill_service.string_of_job_status rejected_job.status) + "failed" ; + [%test_eq: bool] + (Option.value_map rejected_job.error ~default:false ~f:(fun error -> + String.is_substring error ~substring:"active" ) ) + true + +let%test_unit "Backfill progress subscriptions stream GraphQL-SSE events" = + Feature_parser.assert_scenario "backfill-api.feature" + "Backfill progress subscriptions stream GraphQL-SSE events" ; + let service = test_service () in + let job = add_job service () in + let reader = + sse_reader_exn service + (Yojson.Basic.to_string + (`Assoc + [ ( "query" + , `String + (sprintf + {|subscription { backfillProgress(id: "%s") { id status diffsPublished error } }|} + job.id ) ) + ] ) ) + in + let first = pipe_read_string_exn reader |> backfill_progress_event_exn in + assert_basic_json_equal (find_assoc_exn "diffsPublished" first) (`Int 0) ; + Explorer_backfill_service.update_job job ~status:Running ~diffs_published:1 + ~started_at:Time.epoch () ; + let running = pipe_read_string_exn reader |> backfill_progress_event_exn in + assert_basic_json_equal (find_assoc_exn "diffsPublished" running) (`Int 1) ; + Explorer_backfill_service.update_job job ~status:Completed ~diffs_published:2 + ~finished_at:Time.epoch () ; + let completed = pipe_read_string_exn reader |> backfill_progress_event_exn in + assert_basic_json_equal + (find_assoc_exn "status" completed) + (`String "completed") ; + [%test_eq: string] + (pipe_read_string_exn reader) + Explorer_backfill_sse.complete_event + +let%test_unit + "Transaction events expose diff fields and NATS dedup header" = + Feature_parser.assert_scenario "event-contract.feature" + "Transaction events expose diff fields and NATS dedup header" ; + let target_ledger_hash = Ledger_hash.empty_hash in + let diff = sample_diff () in + let diff_json = Da_layer.Diff.Stable.V4.to_yojson diff in + let message = + Explorer_events.build_transaction_message + ~kind:Explorer_events.Transaction_kind.Sync_replay ~target_ledger_hash + ~genesis:false ~diff + in + [%test_eq: string] message.subject Explorer_events.Subject.transactions ; + assert_safe_json_equal + (find_assoc_exn "kind" message.payload) + (`String "sync_replay") ; + assert_safe_json_equal + (find_assoc_exn "target_ledger_hash" message.payload) + (Ledger_hash.to_yojson target_ledger_hash) ; + assert_safe_json_equal + (find_assoc_exn "source_ledger_hash" message.payload) + (find_assoc_exn "source_ledger_hash" diff_json) ; + assert_safe_json_equal + (find_assoc_exn "changed_accounts" message.payload) + (`List []) ; + assert_safe_json_equal (find_assoc_exn "command" message.payload) `Null ; + assert_safe_json_equal (find_assoc_exn "genesis" message.payload) (`Bool false) ; + assert_safe_json_equal + (find_assoc_exn "diff" message.payload) + diff_json ; + [%test_eq: string] + (find_header_exn message.headers "Nats-Msg-Id") + (Ledger_hash.to_decimal_string target_ledger_hash) + +let%test_unit + "Finality events include level, ledger hashes, and NATS dedup header" = + Feature_parser.assert_scenario "event-contract.feature" + "Finality events include level, ledger hashes, and NATS dedup header" ; + let target_ledger_hash = Ledger_hash.empty_hash in + let message = + Explorer_events.build_finality_message ~logger:(Logger.create ()) + ~status:Explorer_events.Finality_status.Proved + ~source_ledger_hash:Ledger_hash.empty_hash ~target_ledger_hash + in + [%test_eq: string] message.subject Explorer_events.Subject.finality ; + assert_safe_json_equal + (find_assoc_exn "level" message.payload) + (`String "proved") ; + assert_safe_json_equal + (find_assoc_exn "ledger_hash" message.payload) + (Ledger_hash.to_yojson target_ledger_hash) ; + assert_safe_json_equal + (find_assoc_exn "source_ledger_hash" message.payload) + (Ledger_hash.to_yojson Ledger_hash.empty_hash) ; + assert_safe_json_equal + (find_assoc_exn "target_ledger_hash" message.payload) + (Ledger_hash.to_yojson target_ledger_hash) ; + [%test_eq: string] + (find_header_exn message.headers "Nats-Msg-Id") + (sprintf "finality-proved-%s" + (Ledger_hash.to_decimal_string target_ledger_hash) ) + +let%test_unit "Health events include the service identity and publishing state" + = + Feature_parser.assert_scenario "event-contract.feature" + "Health events include the service identity and publishing state" ; + let message = + Explorer_events.build_health_message ~logger:(Logger.create ()) + ~service:"sequencer-nats-publisher" ~instance_id:"instance-1" ~status:"ok" + ~last_published_hash:Ledger_hash.empty_hash + ~unproved_hash:Ledger_hash.empty_hash () + in + [%test_eq: string] message.subject Explorer_events.Subject.health ; + assert_safe_json_equal + (find_assoc_exn "service" message.payload) + (`String "sequencer-nats-publisher") ; + assert_safe_json_equal + (find_assoc_exn "instance_id" message.payload) + (`String "instance-1") ; + assert_safe_json_equal (find_assoc_exn "status" message.payload) (`String "ok") ; + assert_safe_json_equal + (find_assoc_exn "last_published_hash" message.payload) + (Ledger_hash.to_yojson Ledger_hash.empty_hash) ; + assert_safe_json_equal + (find_assoc_exn "unproved_hash" message.payload) + (Ledger_hash.to_yojson Ledger_hash.empty_hash) + +let%test_unit "Disabled NATS publishing is a no-op" = + Feature_parser.assert_scenario "event-contract.feature" + "Disabled NATS publishing is a no-op" ; + Explorer_events.publish_transaction Explorer_events.noop_sink + ~kind:Explorer_events.Transaction_kind.User_command + ~target_ledger_hash:Ledger_hash.empty_hash ~genesis:false + ~diff:(sample_diff ()) + +let%test_unit "Backfill dropped publishes are fatal" = + Feature_parser.assert_scenario "event-contract.feature" + "Backfill dropped publishes fail the job" ; + let service = test_service () in + let job = + add_job service ~status:Explorer_backfill_service.Running + ~diffs_published:0 () + in + match + Explorer_backfill_service.handle_backfill_publish_result service job + ~target_ledger_hash:Ledger_hash.empty_hash `Dropped + with + | Ok _ -> + failwith "expected dropped backfill publish to be fatal" + | Error error -> + if + not + (String.is_substring (Error.to_string_hum error) + ~substring:"NATS publish dropped" ) + then + failwithf "unexpected dropped-publish error: %s" + (Error.to_string_hum error) + () ; + [%test_eq: int] job.diffs_published 0 diff --git a/src/app/zeko/sequencer/explorer/tests/explorer_nats_gherkin_tests.ml b/src/app/zeko/sequencer/explorer/tests/explorer_nats_gherkin_tests.ml new file mode 100644 index 0000000000..6dd62d36e3 --- /dev/null +++ b/src/app/zeko/sequencer/explorer/tests/explorer_nats_gherkin_tests.ml @@ -0,0 +1,263 @@ +(* Executes Gherkin-backed explorer integration scenarios against a real NATS + server supplied through the NATS_URL environment variable. *) + +open Core +open Async +open Mina_base + +let url () = + match Sys.getenv "NATS_URL" with + | Some value -> + Uri.of_string value + | None -> + failwith "NATS_URL must be set for explorer NATS integration tests" + +let connect_or_fail uri = + Clock_ns.with_timeout (Time_ns.Span.of_sec 5.) + (Nats_client_async.connect (Some uri)) + >>= function + | `Timeout -> + failwithf "timed out connecting to %s" (Uri.to_string uri) () + | `Result client -> + return client + +let expect_ok label = function + | Ok value -> + value + | Error err -> + failwithf "%s failed: %s" label (Error.to_string_hum err) () + +let read_message label reader = + Clock_ns.with_timeout (Time_ns.Span.of_sec 5.) (Pipe.read reader) + >>= function + | `Timeout -> + failwithf "%s timed out waiting for a message" label () + | `Result `Eof -> + failwithf "%s closed before delivering a message" label () + | `Result (`Ok message) -> + return message + +let wait_for_subscription_registration () = + (* The async NATS client exposes subscribe but not a flush/ack primitive, + so give the server a brief moment to register the SUB before publishing. *) + Clock_ns.after (Time_ns.Span.of_ms 100.) + +let require_header headers ~name ~expected = + match List.Assoc.find headers ~equal:String.equal name with + | Some value when String.equal value expected -> + () + | Some value -> + failwithf "expected header %s=%s but received %s" name expected value () + | None -> + failwithf "missing header %s" name () + +let json_assoc_exn key json = + match json with + | `Assoc fields -> + List.Assoc.find_exn fields key ~equal:String.equal + | _ -> + failwith "Expected JSON object" + +let assert_safe_json_equal actual expected = + if not (Yojson.Safe.equal actual expected) + then + failwithf "Expected %s but got %s" (Yojson.Safe.to_string expected) + (Yojson.Safe.to_string actual) () + +let json_int_exn = function + | `Int value -> + value + | `Intlit value -> + Int.of_string value + | json -> + failwithf "expected JSON integer, got %s" (Yojson.Safe.to_string json) () + +let json_bool_exn = function + | `Bool value -> + value + | json -> + failwithf "expected JSON boolean, got %s" (Yojson.Safe.to_string json) () + +let require_no_jetstream_error label json = + match json with + | `Assoc fields -> ( + match List.Assoc.find fields "error" ~equal:String.equal with + | None -> + () + | Some error -> + failwithf "%s returned JetStream error: %s" label + (Yojson.Safe.to_string error) + () ) + | _ -> + failwithf "%s returned non-object JSON: %s" label + (Yojson.Safe.to_string json) + () + +let jetstream_request_json label client ~operation payload = + let subject = Explorer_events.Jetstream.api_subject operation in + Nats_client_async.request client ~subject ~timeout:(Time_ns.Span.of_sec 5.) + (Yojson.Safe.to_string payload) + >>| fun response -> + let response = expect_ok label response in + let json = Yojson.Safe.from_string response in + require_no_jetstream_error label json ; + json + +let purge_jetstream_stream client = + jetstream_request_json "purge stream" client ~operation:"PURGE" (`Assoc []) + >>| ignore + +let jetstream_message_count client = + jetstream_request_json "stream info" client ~operation:"INFO" (`Assoc []) + >>| fun info -> + json_assoc_exn "messages" (json_assoc_exn "state" info) |> json_int_exn + +let publish_jetstream_message label client + ({ Explorer_events.subject; headers; payload } : Explorer_events.message) = + let headers = Nats_client.Headers.of_list headers in + Nats_client_async.request client ~subject ~headers + ~timeout:(Time_ns.Span.of_sec 5.) + (Yojson.Safe.to_string payload) + >>| fun response -> + let response = expect_ok label response in + let json = Yojson.Safe.from_string response in + require_no_jetstream_error label json ; + json + +let sample_diff ?(source_ledger_hash = Ledger_hash.empty_hash) () = + Explorer_events.build_live_diff ~logger:(Logger.create ()) + ~diff: + (Da_layer.Diff.create_pending ~source_ledger_hash ~changed_accounts:[] + ~actions:(`Actions []) ) + ~acc_set_root:Snark_params.Tick.Field.zero + +let test_service nats_client : Explorer_backfill_service.t = + { logger = Logger.create () + ; da_config = Da_layer.Client.Config.of_string_list [] + ; nats_client = Some nats_client + ; jobs = String.Table.create () + ; instance_id = "integration-instance" + ; started_at = Time.epoch + } + +let with_clients f = + let uri = url () in + let%bind subscriber = connect_or_fail uri in + let%bind actor = connect_or_fail uri in + Monitor.protect + ~finally:(fun () -> + Nats_client_async.close actor >>= fun () -> + Nats_client_async.close subscriber) + (fun () -> f ~subscriber ~actor) + +let run_live_transaction_scenario () = + Feature_parser.assert_scenario "nats-integration.feature" + "Live transaction events round-trip through NATS with the dedup header" ; + with_clients (fun ~subscriber ~actor -> + let%bind subscription = + Nats_client_async.subscribe subscriber + ~subject:Explorer_events.Subject.transactions () + in + let subscription = expect_ok "live transaction subscription" subscription in + let%bind () = wait_for_subscription_registration () in + let target_ledger_hash = Ledger_hash.empty_hash in + let sink = Explorer_events.create_nats_sink actor in + Explorer_events.publish_transaction sink + ~kind:Explorer_events.Transaction_kind.User_command + ~target_ledger_hash ~genesis:false ~diff:(sample_diff ()) ; + let%map message = + read_message "live transaction publish" subscription.messages + in + if not (String.equal message.subject Explorer_events.Subject.transactions) + then + failwithf "unexpected transaction subject: %s" message.subject () ; + let headers = + message.headers + |> Option.value ~default:Nats_client.Headers.empty + |> Nats_client.Headers.to_list + in + require_header headers ~name:"Nats-Msg-Id" + ~expected:(Ledger_hash.to_decimal_string target_ledger_hash) ; + let payload = Yojson.Safe.from_string message.payload in + assert_safe_json_equal (json_assoc_exn "kind" payload) + (`String "user_command") ) + +let run_backfill_scenario () = + Feature_parser.assert_scenario "nats-integration.feature" + "Backfill replay events round-trip through NATS as genesis replays" ; + with_clients (fun ~subscriber ~actor -> + let%bind subscription = + Nats_client_async.subscribe subscriber + ~subject:Explorer_events.Subject.transactions () + in + let subscription = expect_ok "backfill subscription" subscription in + let%bind () = wait_for_subscription_registration () in + let service = test_service actor in + let target_ledger_hash = Ledger_hash.empty_hash in + let publish_result = + Explorer_backfill_service.publish_backfill_diff service + ~from_hash:Explorer_backfill_service.genesis_hash ~index:0 + ~target_ledger_hash + (sample_diff ~source_ledger_hash:Explorer_backfill_service.genesis_hash + () ) + in + if not (Poly.equal publish_result `Queued) + then failwith "expected backfill publish to be queued" ; + let%map message = + read_message "backfill publish" subscription.messages + in + if not (String.equal message.subject Explorer_events.Subject.transactions) + then + failwithf "unexpected backfill subject: %s" message.subject () ; + let headers = + message.headers + |> Option.value ~default:Nats_client.Headers.empty + |> Nats_client.Headers.to_list + in + require_header headers ~name:"Nats-Msg-Id" + ~expected:(Ledger_hash.to_decimal_string target_ledger_hash) ; + let payload = Yojson.Safe.from_string message.payload in + assert_safe_json_equal (json_assoc_exn "kind" payload) + (`String "genesis_replay") ; + assert_safe_json_equal (json_assoc_exn "genesis" payload) (`Bool true) ) + +let run_jetstream_dedupe_scenario () = + Feature_parser.assert_scenario "nats-integration.feature" + "JetStream deduplicates transaction publishes by Nats-Msg-Id" ; + with_clients (fun ~subscriber:_ ~actor -> + let logger = Logger.create () in + let%bind () = Explorer_events.ensure_jetstream_stream ~logger actor in + let%bind () = purge_jetstream_stream actor in + let target_ledger_hash = Ledger_hash.empty_hash in + let message = + Explorer_events.build_transaction_message + ~kind:Explorer_events.Transaction_kind.User_command + ~target_ledger_hash ~genesis:false ~diff:(sample_diff ()) + in + let%bind first_ack = + publish_jetstream_message "first duplicate publish" actor message + in + let%bind second_ack = + publish_jetstream_message "second duplicate publish" actor message + in + assert_safe_json_equal + (json_assoc_exn "stream" first_ack) + (`String Explorer_events.Jetstream.stream_name) ; + assert_safe_json_equal + (json_assoc_exn "stream" second_ack) + (`String Explorer_events.Jetstream.stream_name) ; + if not (json_bool_exn (json_assoc_exn "duplicate" second_ack)) then + failwith "expected the second JetStream publish to be a duplicate" ; + let%map messages = jetstream_message_count actor in + if not (Int.equal messages 1) then + failwithf "expected one stored JetStream message, got %d" messages () ) + +let main () = + let%bind () = run_live_transaction_scenario () in + let%bind () = run_backfill_scenario () in + let%bind () = run_jetstream_dedupe_scenario () in + printf "explorer NATS Gherkin integration scenarios passed against %s\n" + (Uri.to_string (url ())) ; + return () + +let () = Thread_safe.block_on_async_exn main diff --git a/src/app/zeko/sequencer/explorer/tests/feature_parser.ml b/src/app/zeko/sequencer/explorer/tests/feature_parser.ml new file mode 100644 index 0000000000..6f7eab3b45 --- /dev/null +++ b/src/app/zeko/sequencer/explorer/tests/feature_parser.ml @@ -0,0 +1,50 @@ +(* Loads the copied Gherkin feature files and exposes a small parser so the + acceptance tests can stay anchored to the scenario inventory from the spike. *) + +open Core + +type t = + { title : string + ; scenarios : string list + } + +let rec find_workspace_root dir = + if Caml.Sys.file_exists (Filename.concat dir "dune-project") + then Some dir + else + let parent = Filename.dirname dir in + if String.equal parent dir then None else find_workspace_root parent + +let workspace_root () = + let candidates = + [ Caml.Sys.getcwd () + ; Filename.dirname __FILE__ + ; Filename.dirname (Filename.dirname __FILE__) + ] + in + List.find_map candidates ~f:find_workspace_root + |> Option.value_exn ~message:"Could not locate workspace root" + +let feature_path filename = + Filename.concat (workspace_root ()) + ("src/app/zeko/sequencer/explorer/tests/features/" ^ filename) + +let parse_lines lines = + List.fold lines ~init:{ title = ""; scenarios = [] } ~f:(fun acc line -> + match String.chop_prefix ~prefix:"Feature: " (String.strip line) with + | Some title -> + { acc with title } + | None -> ( + match String.chop_prefix ~prefix:"Scenario: " (String.strip line) with + | Some scenario -> + { acc with scenarios = acc.scenarios @ [ scenario ] } + | None -> + acc ) ) + +let load filename = In_channel.read_lines (feature_path filename) |> parse_lines + +let assert_scenario filename scenario = + let feature = load filename in + if not (List.mem feature.scenarios scenario ~equal:String.equal) + then + failwithf "Missing scenario %S in feature file %s" scenario filename () diff --git a/src/app/zeko/sequencer/explorer/tests/features/backfill-api.feature b/src/app/zeko/sequencer/explorer/tests/features/backfill-api.feature new file mode 100644 index 0000000000..6c5cb1a3e4 --- /dev/null +++ b/src/app/zeko/sequencer/explorer/tests/features/backfill-api.feature @@ -0,0 +1,45 @@ +Feature: Explorer Backfill API + + Scenario: A genesis backfill marks only the first replayed diff as genesis + Given the backfill service is replaying from genesis + When it classifies the first replayed diff + Then the diff kind is "genesis_replay" + And later replayed diffs use "sync_replay" + + Scenario: The backfill mutation returns a failed job snapshot for invalid hashes + Given a standalone backfill service + When GraphQL mutation backfill is called with an invalid fromHash + Then the mutation returns a BackfillJob snapshot + And the snapshot status is "failed" + And the snapshot error mentions an invalid ledger hash + + Scenario: Backfill progress subscriptions stream job updates + Given a standalone backfill service with a queued backfill job + When the job publishes progress updates + Then the subscription yields increasing diffsPublished counts + And the final update has status "completed" + + Scenario: The backfill health query exposes the service instance + Given a standalone backfill service + When GraphQL query health is executed + Then the response includes instanceId + And the response includes startedAt + + Scenario: The backfill job query returns the current job snapshot + Given a standalone backfill service with a stored backfill job + When GraphQL query backfillJob is executed for that id + Then the response includes the matching job id + And the response includes the job status + + Scenario: Concurrent backfill requests are bounded + Given a standalone backfill service with an active backfill job + When another backfill is requested for a different range + Then the second backfill request is rejected + And the snapshot error mentions an active job + + Scenario: Backfill progress subscriptions stream GraphQL-SSE events + Given a standalone backfill service with a queued backfill job + When the GraphQL-SSE endpoint subscribes to that job + Then the first SSE event includes diffsPublished 0 + And later SSE events reflect published progress + And the stream finishes with a complete event diff --git a/src/app/zeko/sequencer/explorer/tests/features/e2e.feature b/src/app/zeko/sequencer/explorer/tests/features/e2e.feature new file mode 100644 index 0000000000..8d91c4357e --- /dev/null +++ b/src/app/zeko/sequencer/explorer/tests/features/e2e.feature @@ -0,0 +1,12 @@ +Feature: Explorer End-to-End Publishing + + Scenario: Sequencer-applied transactions are published to NATS + Given the Zeko integration services are running with a NATS broker + When the sequencer applies a user transaction + Then an explorer subscriber receives a "user_command" transaction event + And the transaction event includes the target ledger hash as "Nats-Msg-Id" + + Scenario: Backfill replays sequencer DA diffs to NATS + Given a sequencer-applied transaction has been stored in the DA layer + When the backfill service replays from genesis to that transaction + Then an explorer subscriber receives a replay event for that target ledger hash diff --git a/src/app/zeko/sequencer/explorer/tests/features/event-contract.feature b/src/app/zeko/sequencer/explorer/tests/features/event-contract.feature new file mode 100644 index 0000000000..3f353b20a3 --- /dev/null +++ b/src/app/zeko/sequencer/explorer/tests/features/event-contract.feature @@ -0,0 +1,41 @@ +Feature: Explorer Event Contract + + Scenario: Transaction events expose diff fields and NATS dedup header + Given a transaction diff prepared for explorer publishing + When the transaction event message is built + Then the subject is "zeko.l2.transactions" + And the payload includes "kind" + And the payload includes "source_ledger_hash" + And the payload includes "target_ledger_hash" + And the payload includes "changed_accounts" + And the payload includes "command" + And the payload includes "actions" + And the headers include "Nats-Msg-Id" + + Scenario: Finality events include level, ledger hashes, and NATS dedup header + Given a finality transition prepared for explorer publishing + When the finality event message is built + Then the subject is "zeko.l2.finality" + And the payload includes "level" + And the payload includes "ledger_hash" + And the payload includes "source_ledger_hash" + And the headers include "Nats-Msg-Id" + + Scenario: Health events include the service identity and publishing state + Given sequencer health data prepared for explorer publishing + When the health event message is built + Then the subject is "zeko.health" + And the payload includes "service" + And the payload includes "instance_id" + And the payload includes "last_published_hash" + And the payload includes "unproved_hash" + + Scenario: Disabled NATS publishing is a no-op + Given explorer publishing has no NATS client + When a transaction event is published + Then publishing does not raise + + Scenario: Backfill dropped publishes fail the job + Given a backfill job is publishing historical diffs + When NATS drops the publish + Then the backfill publish is treated as an error diff --git a/src/app/zeko/sequencer/explorer/tests/features/nats-integration.feature b/src/app/zeko/sequencer/explorer/tests/features/nats-integration.feature new file mode 100644 index 0000000000..952aa3c712 --- /dev/null +++ b/src/app/zeko/sequencer/explorer/tests/features/nats-integration.feature @@ -0,0 +1,20 @@ +Feature: Explorer NATS Integration + + Scenario: Live transaction events round-trip through NATS with the dedup header + Given a running NATS server + When the explorer publisher emits a live transaction event + Then a subscriber receives it on "zeko.l2.transactions" + And the message includes the "Nats-Msg-Id" header + And the payload kind is "user_command" + + Scenario: Backfill replay events round-trip through NATS as genesis replays + Given a running NATS server + When the backfill service republishes the first genesis diff + Then a subscriber receives it on "zeko.l2.transactions" + And the payload kind is "genesis_replay" + And the payload marks genesis as true + + Scenario: JetStream deduplicates transaction publishes by Nats-Msg-Id + Given a running NATS server with JetStream enabled + When the same transaction event is published twice with the same "Nats-Msg-Id" + Then the JetStream stream stores one transaction message diff --git a/src/app/zeko/sequencer/explorer/tests/features/sequencer-hooks.feature b/src/app/zeko/sequencer/explorer/tests/features/sequencer-hooks.feature new file mode 100644 index 0000000000..c58e557f2c --- /dev/null +++ b/src/app/zeko/sequencer/explorer/tests/features/sequencer-hooks.feature @@ -0,0 +1,11 @@ +Feature: Sequencer Explorer Hooks + + Scenario: Sync replay from genesis marks the very first diff as genesis + Given the sequencer is replaying diffs from genesis + When it classifies the first replayed diff + Then the replay is marked as genesis + + Scenario: Sync replay from a checkpoint never re-labels diffs as genesis + Given the sequencer is replaying diffs from a checkpoint + When it classifies the first replayed diff + Then the replay is not marked as genesis diff --git a/src/app/zeko/sequencer/explorer/tests/run-explorer-e2e-test.sh b/src/app/zeko/sequencer/explorer/tests/run-explorer-e2e-test.sh new file mode 100755 index 0000000000..d84fa7d8c1 --- /dev/null +++ b/src/app/zeko/sequencer/explorer/tests/run-explorer-e2e-test.sh @@ -0,0 +1,268 @@ +#!/usr/bin/env bash +set -euo pipefail + +usage() { + echo "Usage: $0 [wait_for_port?]" >&2 + exit 1 +} + +if [ "$#" -lt 2 ] || [ "$#" -gt 3 ]; then + usage +fi + +MODE="$1" +NUM_PROVERS="$2" +WAIT_FOR_PORT="${3:-true}" +PROVER_PIDS=() +SIGNER_PIDS=() + +case "$MODE" in +fake | real) ;; +*) + echo "Error: first argument must be 'fake' or 'real'" >&2 + usage + ;; +esac + +if ! [[ "$NUM_PROVERS" =~ ^[1-9][0-9]*$ ]]; then + echo "Error: second argument must be a positive integer" >&2 + usage +fi + +if [ -z "${NATS_URL:-}" ]; then + echo "Error: NATS_URL must point at a running NATS server" >&2 + exit 1 +fi + +cleanup() { + local exit_status=$1 + echo "Cleaning up..." + local -a service_pids=( + "${l1_pid:-}" + "${da1_pid:-}" + "${da2_pid:-}" + "${da3_pid:-}" + "${PROVER_PIDS[@]}" + "${SIGNER_PIDS[@]}" + ) + + trap - SIGINT SIGTERM EXIT + + for pid in "${service_pids[@]}"; do + [ -n "$pid" ] || continue + kill_process_tree TERM "$pid" + done + + sleep 1 + + for pid in "${service_pids[@]}"; do + [ -n "$pid" ] || continue + kill_process_tree KILL "$pid" + done + + [ -z "${TMP_DIR:-}" ] || rm -rf "$TMP_DIR" + docker rm -f pg-sequencer 2>/dev/null + docker rm -f rabbitmq-sequencer 2>/dev/null + exit "${exit_status:-0}" +} + +REPO_ROOT="$(git rev-parse --show-toplevel)" +SEQUENCER_BUILD_ROOT="$REPO_ROOT/_build/default/src/app/zeko/sequencer" +SIGNER_BUILD_ROOT="$REPO_ROOT/_build/default/src/app/zeko/signer" + +export ZEKO_SIGNATURE_KIND=zeko-testnet +export ZEKO_CIRCUITS_CONFIG=test + +TMP_DIR=$(mktemp -d) + +list_child_pids() { + local parent_pid="$1" + ps -axo pid=,ppid= | awk -v parent="$parent_pid" '$2 == parent { print $1 }' +} + +kill_process_tree() { + local signal="$1" + local pid="$2" + local child_pid + + while read -r child_pid; do + [ -n "$child_pid" ] || continue + kill_process_tree "$signal" "$child_pid" + done < <(list_child_pids "$pid") + + kill "-$signal" "$pid" 2>/dev/null || true +} + +trap 'cleanup 1' SIGINT SIGTERM +trap 'cleanup $?' EXIT + +wait_for_port() { + if [ "$WAIT_FOR_PORT" = "true" ]; then + local port=$1 + local pid=$2 + while ! nc -z localhost "$port"; do + sleep 1 + + if ! kill -0 "$pid" 2>/dev/null; then + echo "Process for port $port failed to start" + exit 1 + fi + done + + echo "Port $port is now open" + else + echo "Waiting for 2 seconds..." + sleep 2 + fi +} + +wait_for_queue_consumers() { + local queue=$1 + local expected_consumers=$2 + local timeout_seconds=${3:-300} + + for _ in $(seq 1 "$timeout_seconds"); do + if docker exec rabbitmq-sequencer rabbitmqctl -q list_queues name consumers \ + | awk -v queue="$queue" -v expected="$expected_consumers" \ + '$1 == queue && $2 >= expected { found = 1 } END { exit found ? 0 : 1 }'; then + echo "Queue $queue has at least $expected_consumers consumers" + return + fi + + for pid in "${PROVER_PIDS[@]}"; do + if ! kill -0 "$pid" 2>/dev/null; then + echo "Prover process $pid exited before registering as a queue consumer" + exit 1 + fi + done + + sleep 1 + done + + echo "Timed out waiting for $expected_consumers consumers on queue $queue" + docker logs rabbitmq-sequencer || true + exit 1 +} + +run() { + local name="$1" + shift + + bash -c ' + name="$1" + shift + + exec stdbuf -oL -eL "$@" \ + > >(awk -v n="$name" '"'"'{ print "[" n "] " $0; fflush(); }'"'"') \ + 2> >(awk -v n="$name" '"'"'{ print "[" n "][ERR] " $0; fflush(); }'"'"' >&2) + ' bash "$name" "$@" +} + +KEYGEN_BIN="$SEQUENCER_BUILD_ROOT/cli.exe" + +generate_even_key() { + "$KEYGEN_BIN" generate-even-key | awk -F': ' '/Private key:/ {print $2}' +} + +opam exec -- env -u DUNE_RPC dune build \ + src/app/zeko/sequencer/tests/testing_ledger/run.exe \ + src/app/zeko/da_layer/cli.exe \ + src/app/zeko/sequencer/cli.exe \ + src/app/zeko/sequencer/prover/cli.exe \ + src/app/zeko/sequencer/prover/cli_fake.exe \ + src/app/zeko/signer/cli.exe \ + src/app/zeko/sequencer/explorer/tests/explorer_e2e_gherkin_tests.exe + +docker run --rm --name pg-sequencer \ + -e POSTGRES_USER=postgres \ + -e POSTGRES_PASSWORD=postgres \ + --tmpfs /var/lib/postgresql/data:rw,noexec,nosuid \ + -p 5433:5432 \ + -d postgres:16-alpine + +docker run -d --name rabbitmq-sequencer \ + -p 5672:5672 \ + rabbitmq:3.13.7 + +wait_for_port 5433 $$ +wait_for_port 5672 $$ + +SEQUENCER_SIGNER_BIN="$SIGNER_BUILD_ROOT/cli.exe" +DA_SIGNER_BIN="$SIGNER_BUILD_ROOT/cli.exe" + +ZEKO_TEST_SEQUENCER_SIGNER_PRIVATE_KEY="$(generate_even_key)" +DA1_SIGNER_PRIVATE_KEY="$(generate_even_key)" +DA2_SIGNER_PRIVATE_KEY="$(generate_even_key)" +DA3_SIGNER_PRIVATE_KEY="$(generate_even_key)" + +export ZEKO_TEST_SEQUENCER_SIGNER="127.0.0.1:8600" +export ZEKO_TEST_SEQUENCER_SIGNER_PRIVATE_KEY + +run "sequencer-signer" \ + env MINA_PRIVATE_KEY="$ZEKO_TEST_SEQUENCER_SIGNER_PRIVATE_KEY" \ + "$SEQUENCER_SIGNER_BIN" run --port 8600 --allow-zkapp-signing --max-fee 10 --max-balance-change 1000000 & +signer_seq_pid=$! +SIGNER_PIDS+=("$signer_seq_pid") + +run "da1-signer" \ + env MINA_PRIVATE_KEY="$DA1_SIGNER_PRIVATE_KEY" \ + "$DA_SIGNER_BIN" run --port 8601 --allow-field-signing & +signer_da1_pid=$! +SIGNER_PIDS+=("$signer_da1_pid") + +run "da2-signer" \ + env MINA_PRIVATE_KEY="$DA2_SIGNER_PRIVATE_KEY" \ + "$DA_SIGNER_BIN" run --port 8602 --allow-field-signing & +signer_da2_pid=$! +SIGNER_PIDS+=("$signer_da2_pid") + +run "da3-signer" \ + env MINA_PRIVATE_KEY="$DA3_SIGNER_PRIVATE_KEY" \ + "$DA_SIGNER_BIN" run --port 8603 --allow-field-signing & +signer_da3_pid=$! +SIGNER_PIDS+=("$signer_da3_pid") + +wait_for_port 8600 "$signer_seq_pid" +wait_for_port 8601 "$signer_da1_pid" +wait_for_port 8602 "$signer_da2_pid" +wait_for_port 8603 "$signer_da3_pid" + +run "l1" "$SEQUENCER_BUILD_ROOT/tests/testing_ledger/run.exe" -p 8080 --db-dir "$TMP_DIR/l1_db" --network-id mainnet --block-period 9999999 & +l1_pid=$! + +run "da1" "$SEQUENCER_BUILD_ROOT/../da_layer/cli.exe" run-node --port 8555 --healthcheck-port 8558 --network-id zeko-testnet --db-dir "$TMP_DIR/da1_db" --signer 127.0.0.1:8601 & +da1_pid=$! + +run "da2" "$SEQUENCER_BUILD_ROOT/../da_layer/cli.exe" run-node --port 8556 --healthcheck-port 8559 --network-id zeko-testnet --db-dir "$TMP_DIR/da2_db" --signer 127.0.0.1:8602 & +da2_pid=$! + +run "da3" "$SEQUENCER_BUILD_ROOT/../da_layer/cli.exe" run-node --port 8557 --healthcheck-port 8560 --network-id zeko-testnet --db-dir "$TMP_DIR/da3_db" --signer 127.0.0.1:8603 & +da3_pid=$! + +if [ "$MODE" = "fake" ]; then + sleep 5 + PROVER_BIN="$SEQUENCER_BUILD_ROOT/prover/cli_fake.exe" +else + PROVER_BIN="$SEQUENCER_BUILD_ROOT/prover/cli.exe" +fi + +for ((i = 0; i < NUM_PROVERS; i++)); do + run "prover-$i" \ + env ZEKO_CIRCUITS_MODE="$MODE" \ + "$PROVER_BIN" run-server --mq-host "localhost:5672" & + PROVER_PIDS+=("$!") +done + +echo "Waiting for services to start..." + +wait_for_port 8080 "$l1_pid" +wait_for_port 8555 "$da1_pid" +wait_for_port 8556 "$da2_pid" +wait_for_port 8557 "$da3_pid" +wait_for_queue_consumers sequencer.jobs "$NUM_PROVERS" + +echo "All services started successfully" + +run "explorer-e2e" \ + env ZEKO_CIRCUITS_MODE="$MODE" \ + "$SEQUENCER_BUILD_ROOT/explorer/tests/explorer_e2e_gherkin_tests.exe" diff --git a/src/app/zeko/sequencer/lib/dune b/src/app/zeko/sequencer/lib/dune index 03c6e946f3..8540ae8239 100644 --- a/src/app/zeko/sequencer/lib/dune +++ b/src/app/zeko/sequencer/lib/dune @@ -12,6 +12,9 @@ indexed_merkle_tree utils gql_client + explorer_events + nats-client + nats-client-async ;; mina ;; mina_base mina_base.import diff --git a/src/app/zeko/sequencer/lib/zeko_sequencer.ml b/src/app/zeko/sequencer/lib/zeko_sequencer.ml index ae0c236cec..813c98673f 100644 --- a/src/app/zeko/sequencer/lib/zeko_sequencer.ml +++ b/src/app/zeko/sequencer/lib/zeko_sequencer.ml @@ -72,6 +72,7 @@ module Sequencer = struct type t = { provers : Zeko_prover.Client.t ; da_client : Da_layer.Client.t + ; nats_sink : Explorer_events.sink ; executor : Executor.t ; config : Config.t ; sequencer_state : State.t @@ -135,6 +136,7 @@ module Sequencer = struct let process ({ da_client + ; nats_sink ; provers ; executor ; config @@ -182,17 +184,34 @@ module Sequencer = struct let%bind _hash = Executor.send_zkapp_command ~logger executor command in + let source_ledger_hash = + Sparse_ledger.merkle_root old_inner_ledger + in + let target_ledger_hash = + Sparse_ledger.merkle_root new_inner_ledger + in State.Last_committed_ledger.set sequencer_state ~data:new_inner_ledger ; + let () = + [%log debug] + "Publishing explorer finality event: subject=%s status=%s \ + source_ledger_hash=%s target_ledger_hash=%s" + Explorer_events.Subject.finality + (Explorer_events.Finality_status.to_string + Explorer_events.Finality_status.Committed ) + (Ledger_hash.to_decimal_string source_ledger_hash) + (Ledger_hash.to_decimal_string target_ledger_hash) ; + Explorer_events.publish_finality nats_sink ~logger + ~status:Explorer_events.Finality_status.Committed + ~source_ledger_hash ~target_ledger_hash + in return (fun () -> let open Relational_db in Pool.use (fun conn -> Committer.Commit_table.insert conn - { source_ledger_hash = - Sparse_ledger.merkle_root old_inner_ledger - ; target_ledger_hash = - Sparse_ledger.merkle_root new_inner_ledger + { source_ledger_hash + ; target_ledger_hash ; witness = commit_witness } ) db_pool ) ) @@ -236,6 +255,9 @@ module Sequencer = struct ; merger : Merger.M.t ; merger_ctx : Merger.Context.t ; da_client : Da_layer.Client.t + ; nats_client : Nats_client_async.client option + ; nats_sink : Explorer_events.sink + ; instance_id : string ; closed : unit Ivar.t ; apply_q : unit Sequencer.t (* Applying of the user command is async operation, but we need to keep the application synchronous *) @@ -245,6 +267,13 @@ module Sequencer = struct let logger = t.logger in [%log info] "Shutting down sequencer" ; Ivar.fill t.closed () ; + let%bind () = + match t.nats_client with + | None -> + return () + | Some client -> + Nats_client_async.close client + in Da_layer.Client.stop t.da_client ; L.Db.close t.ledger ; Indexed_merkle_tree.Db.close t.imt ; @@ -279,6 +308,48 @@ module Sequencer = struct ; committed_ledger_hash = Field.zero } + let publish_transaction_event t ~kind ~target_ledger_hash ~genesis ~diff = + let logger = t.logger in + [%log debug] + "Publishing explorer transaction event: subject=%s kind=%s \ + target_ledger_hash=%s genesis=%b" + Explorer_events.Subject.transactions + (Explorer_events.Transaction_kind.to_string kind) + (Ledger_hash.to_decimal_string target_ledger_hash) + genesis ; + Explorer_events.publish_transaction t.nats_sink ~kind + ~target_ledger_hash ~genesis ~diff + + let publish_finality_event t ~status ~source_ledger_hash ~target_ledger_hash = + let logger = t.logger in + [%log debug] + "Publishing explorer finality event: subject=%s status=%s \ + source_ledger_hash=%s target_ledger_hash=%s" + Explorer_events.Subject.finality + (Explorer_events.Finality_status.to_string status) + (Ledger_hash.to_decimal_string source_ledger_hash) + (Ledger_hash.to_decimal_string target_ledger_hash) ; + Explorer_events.publish_finality t.nats_sink ~logger:t.logger ~status + ~source_ledger_hash ~target_ledger_hash + + let publish_health_event t = + let logger = t.logger in + let { State_hashes.unproved_ledger_hash; _ } = get_latest_state t in + [%log debug] + "Publishing explorer health event: subject=%s service=%s \ + last_published_hash=%s unproved_hash=%s" + Explorer_events.Subject.health "sequencer-nats-publisher" + (Ledger_hash.to_decimal_string (get_root t)) + (Ledger_hash.to_decimal_string unproved_ledger_hash) ; + Explorer_events.publish_health t.nats_sink ~logger:t.logger + ~service:"sequencer-nats-publisher" ~instance_id:t.instance_id + ~status:"ok" ~last_published_hash:(get_root t) + ~unproved_hash:unproved_ledger_hash () + + let diff_with_metadata ~logger ~diff ~acc_set_openings = + Da_layer.Diff.add_time_and_acc_set ~logger diff + ~acc_set:(Indexed_merkle_tree.Sparse.merkle_root acc_set_openings) + let apply_events_and_actions ledger archive command = let ledger = L.of_database ledger in Zkapp_command.(Call_forest.to_list (Poly.account_updates command)) @@ -512,14 +583,24 @@ module Sequencer = struct Account_id.derive_token_id ~owner:(Account.identifier account) ) in + let acc_set_openings = + Indexed_merkle_tree.Sparse.of_db_subset ~logger:t.logger ~db:t.imt + ~keys:new_accounts_keys + in + let target_ledger_hash = L.Db.merkle_root t.ledger in + let published_diff = + diff_with_metadata ~logger:t.logger ~diff ~acc_set_openings + in let%bind () = Da_layer.Client.enqueue_diff t.da_client ~genesis:false - ~ledger_openings:source_ledger - ~acc_set_openings: - (Indexed_merkle_tree.Sparse.of_db_subset ~logger:t.logger - ~db:t.imt ~keys:new_accounts_keys ) - ~diff - ~target_ledger_hash:(L.Db.merkle_root t.ledger) + ~ledger_openings:source_ledger ~acc_set_openings ~diff + ~target_ledger_hash + in + let () = + publish_transaction_event t + ~kind:Explorer_events.Transaction_kind.User_command + ~target_ledger_hash + ~genesis:false ~diff:published_diff in (* Add witnesses to the merger *) @@ -592,14 +673,23 @@ module Sequencer = struct Account_id.derive_token_id ~owner:(Account.identifier account) ) in + let acc_set_openings = + Indexed_merkle_tree.Sparse.of_db_subset ~logger:t.logger + ~db:t.imt ~keys:new_accounts_keys + in + let target_ledger_hash = L.Db.merkle_root t.ledger in + let published_diff = + diff_with_metadata ~logger:t.logger ~diff ~acc_set_openings + in let%bind () = Da_layer.Client.enqueue_diff t.da_client ~genesis:false - ~ledger_openings:source_ledger - ~acc_set_openings: - (Indexed_merkle_tree.Sparse.of_db_subset ~logger:t.logger - ~db:t.imt ~keys:new_accounts_keys ) - ~diff - ~target_ledger_hash:(L.Db.merkle_root t.ledger) + ~ledger_openings:source_ledger ~acc_set_openings ~diff + ~target_ledger_hash + in + let () = + publish_transaction_event t + ~kind:Explorer_events.Transaction_kind.Fee_transfer + ~target_ledger_hash ~genesis:false ~diff:published_diff in match%map Merger.P.add_job t.db_pool t.merger t.merger_ctx ~data:witness @@ -775,6 +865,12 @@ module Sequencer = struct let%bind () = match%map ledger_applied >>| Or_error.ok_exn with | Some (stmt, _) -> + let () = + publish_finality_event t + ~status:Explorer_events.Finality_status.Proved + ~source_ledger_hash:stmt.source_ledger + ~target_ledger_hash:stmt.target_ledger + in [%log info] "Committed: %s -> %s" (Ledger_hash.to_decimal_string stmt.source_ledger) (Ledger_hash.to_decimal_string stmt.target_ledger) @@ -786,7 +882,26 @@ module Sequencer = struct in don't_wait_for (within' ~monitor:Monitor.main (fun () -> go ())) - let sync ~logger ({ config; _ } as t) da_config source = + let run_health_heartbeat t = + let period = Time_ns.Span.of_sec 30. in + let rec go () = + let () = publish_health_event t in + let%bind () = Deferred.any [ after period; Ivar.read t.closed ] in + if Ivar.is_full t.closed then return () else go () + in + don't_wait_for (within' ~monitor:Monitor.main (fun () -> go ())) + + let replay_genesis_flag + ~(source : [ `Genesis | `Specific of Field.t ]) + ~current_chunk ~current_diff = + match source with + | `Genesis -> + current_chunk = 0 && current_diff = 0 + | `Specific _ -> + false + + let sync ~logger ({ config; _ } as t) da_config + (source : [ `Genesis | `Specific of Field.t ]) = [%log info] "Syncing" ; let%bind commited_ledger_hash = Gql_client.infer_state ~logger config.l1_uri @@ -829,11 +944,20 @@ module Sequencer = struct , `Target target_ledger_hash ) -> let genesis = i = 0 in + let published_diff = + diff_with_metadata ~logger:t.logger ~diff ~acc_set_openings + in let%map () = Da_layer.Client.enqueue_diff t.da_client ~diff ~ledger_openings ~acc_set_openings ~target_ledger_hash ~genesis in + publish_transaction_event t + ~kind: + (if genesis then + Explorer_events.Transaction_kind.Genesis_replay + else Explorer_events.Transaction_kind.Sync_replay) + ~target_ledger_hash ~genesis ~diff:published_diff ; genesis || acc ) in [%log info] "Enqueued genesis diff" ; @@ -884,15 +1008,31 @@ module Sequencer = struct Da_layer.Client.get_acc_set_openings ~logger ~diff ~ledger_openings ~imt:t.imt in + let target_ledger_hash = L.Db.merkle_root t.ledger in + let genesis = + (not inserted_genesis) + && replay_genesis_flag ~source ~current_chunk ~current_diff + in + let published_diff = + diff_with_metadata ~logger:t.logger + ~diff:(Da_layer.Diff.drop_time diff) + ~acc_set_openings + in (* Store diff to DA client *) let%bind () = Da_layer.Client.enqueue_diff t.da_client ~diff:(Da_layer.Diff.drop_time diff) ~ledger_openings ~acc_set_openings - ~target_ledger_hash:(L.Db.merkle_root t.ledger) - ~genesis: - ((not inserted_genesis) && current_chunk = 0 && current_diff = 0) + ~target_ledger_hash ~genesis + in + let () = + publish_transaction_event t + ~kind: + (if genesis then + Explorer_events.Transaction_kind.Genesis_replay + else Explorer_events.Transaction_kind.Sync_replay) + ~target_ledger_hash ~genesis ~diff:published_diff in (* Add events and actions *) @@ -1030,11 +1170,12 @@ module Sequencer = struct | true, false | false, true -> failwithf "Corrupted db %s and %s" ledger_dir imt_dir () - let create ~logger ~max_pool_size ~commitment_period_sec ~da_config ~da_keys + let create ?nats_url ~logger ~max_pool_size ~commitment_period_sec ~da_config + ~da_keys ~da_quorum ~db_dir ~checkpoints_dir ~postgres_uri ~l1_uri ~archive_uri ~(signer : Signer_service.Signer.t) ~deposit_delay_blocks ~mq_host ~fee_modifier ~minimum_fee ~slot_acceptance ~proof_cache_db ~l1_config - ~commit_validity_period ~commit_fee ~bridge_txn_fee = + ~commit_validity_period ~commit_fee ~bridge_txn_fee () = [%log info] "Precomputing srs" ; Pickles.Side_loaded.srs_precomputation () ; let db_dir = @@ -1074,6 +1215,20 @@ module Sequencer = struct Da_layer.Client.create ~logger ~config:da_config ~quorum:da_quorum ~da_keys ~db_pool in + let%bind nats_client = + match nats_url with + | None -> + return None + | Some uri -> + Explorer_events.connect_and_ensure_jetstream ~logger uri + in + let nats_sink = + match nats_client with + | None -> + Explorer_events.noop_sink + | Some client -> + Explorer_events.create_nats_sink ~logger client + in let kvdb = L.Db.zeko_kvdb ledger in let%bind provers = Zeko_prover.Client.create ~logger ~db_pool ~mq_host in let executor = @@ -1085,6 +1240,7 @@ module Sequencer = struct Merger.Context. { provers ; da_client + ; nats_sink ; executor ; config ; sequencer_state = kvdb @@ -1225,6 +1381,9 @@ module Sequencer = struct ; archive ; config ; da_client + ; nats_client + ; nats_sink + ; instance_id = Uuid.to_string (Uuid_unix.create ()) ; bridge_prover ; merger ; merger_ctx @@ -1251,5 +1410,6 @@ module Sequencer = struct let () = Da_layer.Client.start_client da_client ~target_ledger_hash:(get_root t) in + let () = run_health_heartbeat t in return t end diff --git a/src/app/zeko/sequencer/run.ml b/src/app/zeko/sequencer/run.ml index d970779650..9987d346af 100644 --- a/src/app/zeko/sequencer/run.ml +++ b/src/app/zeko/sequencer/run.ml @@ -9,7 +9,7 @@ module Sequencer = Zeko_sequencer.Sequencer let run ~logger ~port ~max_pool_size ~commitment_period ~da_config ~da_keys ~da_quorum ~db_dir ~checkpoints_dir ~postgres_uri ~l1_uri ~archive_uri - ~signer ~deposit_delay_blocks ~mq_host ~fee_modifier ~minimum_fee + ~signer ~deposit_delay_blocks ~mq_host ~fee_modifier ~minimum_fee ~nats_url ~slot_acceptance ~commit_validity_period ~commit_fee ~bridge_txn_fee () = let proof_cache_db = Proof_cache_tag.create_identity_db () in let l1_config : Utils.Slot.l1_config = @@ -37,9 +37,10 @@ let run ~logger ~port ~max_pool_size ~commitment_period ~da_config ~da_keys Sequencer.create ~logger ~max_pool_size ~da_config ~da_keys ~da_quorum ~db_dir:(Some db_dir) ~checkpoints_dir:(Some checkpoints_dir) ~postgres_uri ~l1_uri ~archive_uri - ~commitment_period_sec:commitment_period ~deposit_delay_blocks ~signer + ~commitment_period_sec:commitment_period ~deposit_delay_blocks + ?nats_url ~signer ~mq_host ~fee_modifier ~minimum_fee ~slot_acceptance ~proof_cache_db - ~l1_config ~commit_validity_period ~commit_fee ~bridge_txn_fee ) + ~l1_config ~commit_validity_period ~commit_fee ~bridge_txn_fee () ) in Sequencer.run_committer sequencer ; @@ -128,6 +129,9 @@ let () = flag "--minimum-fee" (optional_with_default 0.01 float) ~doc:"float Minimum fee for the sequencer" + and nats_url = + flag "--nats-url" (optional string) + ~doc:"string Optional NATS URL for explorer event publishing" and slot_acceptance_m = flag "--slot-acceptance" (optional_with_default 60. float) @@ -155,6 +159,7 @@ let () = in let l1_uri = Uri.of_string l1_uri in let archive_uri = Uri.of_string archive_uri in + let nats_url = Option.map nats_url ~f:Uri.of_string in let mq_host = Host_and_port.of_string mq_host in let logger = Logger.create () in let postgres_uri = Uri.of_string postgres_uri in @@ -171,5 +176,6 @@ let () = run ~logger ~port ~max_pool_size ~commitment_period ~da_config ~da_keys ~da_quorum ~db_dir ~checkpoints_dir ~postgres_uri ~l1_uri ~archive_uri ~signer ~deposit_delay_blocks ~mq_host ~fee_modifier ~minimum_fee - ~slot_acceptance ~commit_validity_period ~commit_fee ~bridge_txn_fee ) + ~nats_url ~slot_acceptance ~commit_validity_period ~commit_fee + ~bridge_txn_fee ) |> Command_unix.run diff --git a/src/app/zeko/sequencer/tests/run-sequencer-test.sh b/src/app/zeko/sequencer/tests/run-sequencer-test.sh index 2f35fdfdb9..6ee2ea8956 100755 --- a/src/app/zeko/sequencer/tests/run-sequencer-test.sh +++ b/src/app/zeko/sequencer/tests/run-sequencer-test.sh @@ -63,7 +63,7 @@ cleanup() { kill_process_tree KILL "$pid" done - rm -rf "$TMP_DIR" + [ -z "${TMP_DIR:-}" ] || rm -rf "$TMP_DIR" docker rm -f pg-sequencer 2>/dev/null docker rm -f rabbitmq-sequencer 2>/dev/null exit ${exit_status:-0} @@ -119,6 +119,34 @@ wait_for_port() { fi } +wait_for_queue_consumers() { + local queue=$1 + local expected_consumers=$2 + local timeout_seconds=${3:-300} + + for _ in $(seq 1 "$timeout_seconds"); do + if docker exec rabbitmq-sequencer rabbitmqctl -q list_queues name consumers \ + | awk -v queue="$queue" -v expected="$expected_consumers" \ + '$1 == queue && $2 >= expected { found = 1 } END { exit found ? 0 : 1 }'; then + echo "Queue $queue has at least $expected_consumers consumers" + return + fi + + for pid in "${PROVER_PIDS[@]}"; do + if ! kill -0 "$pid" 2>/dev/null; then + echo "Prover process $pid exited before registering as a queue consumer" + exit 1 + fi + done + + sleep 1 + done + + echo "Timed out waiting for $expected_consumers consumers on queue $queue" + docker logs rabbitmq-sequencer || true + exit 1 +} + KEYGEN_BIN="$SEQUENCER_BUILD_ROOT/cli.exe" generate_even_key() { @@ -134,7 +162,7 @@ docker run --rm --name pg-sequencer \ docker run -d --name rabbitmq-sequencer \ -p 5672:5672 \ - rabbitmq:latest + rabbitmq:3.13.7 wait_for_port 5433 $$ wait_for_port 5672 $$ @@ -230,6 +258,7 @@ wait_for_port 8080 $l1_pid wait_for_port 8555 $da1_pid wait_for_port 8556 $da2_pid wait_for_port 8557 $da3_pid +wait_for_queue_consumers sequencer.jobs "$NUM_PROVERS" echo "All services started successfully" diff --git a/src/app/zeko/sequencer/tests/sequencer_test.ml b/src/app/zeko/sequencer/tests/sequencer_test.ml index db4b4f1ba0..5301cf8a40 100644 --- a/src/app/zeko/sequencer/tests/sequencer_test.ml +++ b/src/app/zeko/sequencer/tests/sequencer_test.ml @@ -240,6 +240,7 @@ let () = (Mina_numbers.Global_slot_span.of_int 10) ~commit_fee:(Currency.Fee.of_mina_int_exn 1) ~bridge_txn_fee:(Currency.Fee.of_mina_string_exn "0.1") + () in [%test_eq: Frozen_ledger_hash.t] (get_root new_sequencer) final_ledger_hash ; @@ -343,7 +344,8 @@ let () = ~l1_config ~commit_validity_period:(Mina_numbers.Global_slot_span.of_int 10) ~commit_fee:(Currency.Fee.of_mina_int_exn 1) - ~bridge_txn_fee:(Currency.Fee.of_mina_string_exn "0.1") ) + ~bridge_txn_fee:(Currency.Fee.of_mina_string_exn "0.1") + () ) in print_endline "(* Requeue witnesses and commit with quorum 3 *)" ; @@ -475,7 +477,8 @@ let () = ~l1_config ~commit_validity_period:(Mina_numbers.Global_slot_span.of_int 10) ~commit_fee:(Currency.Fee.of_mina_int_exn 1) - ~bridge_txn_fee:(Currency.Fee.of_mina_string_exn "0.1") ) + ~bridge_txn_fee:(Currency.Fee.of_mina_string_exn "0.1") + () ) in print_endline "(* Check that all da nodes are synced *)" ; @@ -583,7 +586,8 @@ let () = ~l1_config ~commit_validity_period:(Mina_numbers.Global_slot_span.of_int 10) ~commit_fee:(Currency.Fee.of_mina_int_exn 1) - ~bridge_txn_fee:(Currency.Fee.of_mina_string_exn "0.1") ) + ~bridge_txn_fee:(Currency.Fee.of_mina_string_exn "0.1") + () ) in print_endline "(* Check that after restart it recommited *)" ; diff --git a/src/app/zeko/sequencer/tests/test_spec.ml b/src/app/zeko/sequencer/tests/test_spec.ml index 65c920282a..70318464a8 100644 --- a/src/app/zeko/sequencer/tests/test_spec.ml +++ b/src/app/zeko/sequencer/tests/test_spec.ml @@ -389,7 +389,7 @@ module Sequencer_spec = struct ; l2_executor : Executor.t } - let gen ?(delay_deposit = 0) ?(number_of_transactions = 5) ?db_dir + let gen ?(delay_deposit = 0) ?(number_of_transactions = 5) ?db_dir ?nats_url ?checkpoints_dir ?(commit_validity_period = Global_slot_span.of_int 10) ~logger ~postgres_uri ~gql_uri ~da_config ~da_keys ~da_quorum ~mq_host ~slot_acceptance () = @@ -556,14 +556,16 @@ module Sequencer_spec = struct let sequencer = run (fun () -> - Sequencer.create ~logger ~max_pool_size:10 ~commitment_period_sec:0. + Sequencer.create ?nats_url ~logger ~max_pool_size:10 + ~commitment_period_sec:0. ~da_config ~da_keys ~da_quorum ~db_dir ~postgres_uri ~l1_uri:gql_uri ~archive_uri:gql_uri ~signer ~deposit_delay_blocks:delay_deposit ~mq_host ~fee_modifier:1.0 ~minimum_fee:0.01 ~slot_acceptance ~proof_cache_db:(Proof_cache_tag.create_identity_db ()) ~l1_config ~commit_validity_period ~checkpoints_dir ~commit_fee:(Currency.Fee.of_mina_int_exn 1) - ~bridge_txn_fee:(Currency.Fee.of_mina_string_exn "0.1") ) + ~bridge_txn_fee:(Currency.Fee.of_mina_string_exn "0.1") + () ) in let l1_executor = Executor.create ~kind:(`L1 gql_uri)