diff --git a/docs/architecture.md b/docs/architecture.md index c36ae26..e82358f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -92,10 +92,12 @@ All of the server's input and output artifacts live in one directory, the *io di | `receipts/receipt_.json` | persistent | the semantics (on success) or the server (on failure) | one stored receipt per transaction, keyed by tx hash, answering `getTransaction`. Each is `{status, ledger, createdAt, envelopeXdr, resultXdr, resultMetaXdr}`. | | `traces/trace_.jsonl` | persistent | the semantics | one execution trace per transaction, keyed by tx hash — the instruction-level records, one JSON object per line. `traceTransaction` returns this file's contents. | | `requests/request_.json` | persistent | the server | an archive of each incoming JSON-RPC request, numbered by a monotonic counter, kept for debugging. | +| `events/events_.json` | persistent | the server | one JSON array per ledger with the contract events emitted in it, already in the spec's Event shape. Built by the server from the staged records after a successful transaction; read back by the K `getEvents` rules. | | `request.json` | transient | the server | the request envelope for the call in flight (`method`, `id`, `now`, and method-specific fields). The semantics remove it once they respond. | | `response.json` | transient | the semantics | the JSON-RPC response (`{jsonrpc, id, result}`) for the most recent call. The server reads it back; it is absent when a transaction gets stuck. | +| `events_staged.jsonl` | transient | the K semantics | the contract events of the transaction in flight, one JSON record per `contract_event` call. The server converts them into `events/events_.json` after a successful run and removes the file. | -Receipts, traces, and request archives are split into one file per item — keyed by tx hash, or numbered — so that no single file grows without bound as the chain advances. The server creates the `receipts/`, `traces/`, and `requests/` directories before the semantics run, because the K file-system hooks open files with POSIX `open()`, which does not create parent directories. +Receipts, traces, events, and request archives are split into one file per item — keyed by tx hash or ledger, or numbered — so that no single file grows without bound as the chain advances. The server creates the `receipts/`, `traces/`, `requests/`, and `events/` directories before the semantics run, because the K file-system hooks open files with POSIX `open()`, which does not create parent directories. The world state stays in KORE (rather than a JSON snapshot) because an uploaded wasm module is a `ModuleDecl` that the semantics cannot reconstruct from bytes — only `wasm2kast` (Python) can produce it. The receipts and the ledger counter, by contrast, are plain data and live in JSON files, which the semantics read and write directly via the file-system hooks. @@ -177,6 +179,7 @@ sequenceDiagram - `resultXdr` / `resultMetaXdr` in `getTransaction` responses (contract return values) - `simulateTransaction` (dry-run without state mutation) -- `getEvents`, `getLedgerEntries`, `getFeeStats` and other read-only RPC methods +- `getLedgerEntries`, `getFeeStats` and other read-only RPC methods +- `system` events in `getEvents` (the semantics have no source of them) - `ExtendFootprintTTL` and `RestoreFootprint` operations - `SCVec` / `SCMap` contract-argument types in the request encoder (`scval_to_json`) diff --git a/docs/node-semantics.md b/docs/node-semantics.md index 6f72b79..051d342 100644 --- a/docs/node-semantics.md +++ b/docs/node-semantics.md @@ -17,6 +17,8 @@ The semantics communicate with the Python process through files in the working d | `metadata.json` | K ↔ K | `{"latest_ledger": N}` — the ledger counter | | `receipts/receipt_.json` | K → Python | one stored receipt per transaction, keyed by tx hash | | `traces/trace_.jsonl` | K → Python | one execution trace per transaction (per-instruction records), keyed by tx hash | +| `events_staged.jsonl` | K → Python | the contract events of the transaction in flight, one JSON record per `contract_event` call | +| `events/events_.json` | Python → K | one finished JSON array of Event objects per ledger, served by `getEvents` | --- @@ -39,7 +41,7 @@ insert-handleRequestFile → handleRequestFile ▼ #dispatchMethod(method, request) ← routes on the "method" field │ - ├─ getHealth / getNetwork / getLatestLedger / getTransaction / traceTransaction → #respond(...) + ├─ getHealth / getNetwork / getLatestLedger / getTransaction / traceTransaction / getEvents → #respond(...) │ └─ sendTransaction → #runTx → run steps → #finalizeTx → record receipt + bump ledger → #respond(...) @@ -62,8 +64,9 @@ If `request.json` is absent, `insert-handleRequestFile` does not fire and K halt - `getNetwork` → `{ "friendbotUrl": null, "passphrase": ..., "protocolVersion": ... }` (passphrase/version come from the request, keeping the semantics network-agnostic) - `getLatestLedger` → reads `metadata.json` and returns `{ "id": <64 zeros>, "protocolVersion": ..., "sequence": }` - `getTransaction` → reads the hash's `receipts/receipt_.json` file; returns the stored receipt merged with the current `latestLedger`/`latestLedgerCloseTime`, or `{ "status": "NOT_FOUND", ... }` when the file is absent +- `getEvents` → scans the `events/events_.json` files of the requested window, applies the request's filters (type, contract ids, topic matchers with `*`/`**`), and paginates; returns `{ "latestLedger": ..., "events": [...], "cursor": ... }`. The events themselves were captured during `sendTransaction`: a rule in `node.md` shadows the upstream no-op `contract_event` host function, resolves the topics/data host objects, and stages them in `events_staged.jsonl`; the Python server then produces the finished per-ledger files (base64 SCVal XDR and strkey ids are XDR work K cannot do). A `startLedger` beyond the chain tip is answered with `#respondError` (JSON-RPC `-32600`). -`#respond(ID, RESULT)` is the shared terminal: it writes the JSON-RPC envelope to `response.json`, removes `request.json`, and sets the exit code to 0. +`#respond(ID, RESULT)` is the shared terminal: it writes the JSON-RPC envelope to `response.json`, removes `request.json`, and sets the exit code to 0. `#respondError(ID, CODE, MESSAGE)` is its error-shaped counterpart, writing `{jsonrpc, id, error: {code, message}}`. --- diff --git a/docs/notes.md b/docs/notes.md index f1a0b3d..3568154 100644 --- a/docs/notes.md +++ b/docs/notes.md @@ -23,6 +23,7 @@ State lives in the io dir as `state.kore` (KORE world state) and `metadata.json` ## Tests (`src/tests/integration/`) - `test_server.py` drives the running HTTP server end-to-end. It exercises the read-only methods, `sendTransaction` + `getTransaction`, ledger increments, the full lifecycle (create → upload wasm → deploy → invoke), and the `traceTransaction` flows. `test_call_tx_with_args` deploys `args.wat` and calls functions with `bool`, `u32`, `i32`, `u64`, `i64`, `u128`, `i128`, and `symbol` arguments, exercising the `scval_to_json` / `#decodeArg` pipeline. +- `test_get_events.py` covers `getEvents`: request validation, the empty-window shape, the full shape of an event emitted via `contract_event` (`data/wasm/events.wat`), filtering, and pagination. - `test_integration.py` and `test_unit.py` hold small sanity checks. Run with `make test` (requires `make kdist-build` first). @@ -35,4 +36,5 @@ The tests do not yet cover `bytes` / `address` SCVal arguments or `SCVec` / `SCM - `resultXdr` / `resultMetaXdr` are empty stubs (contract return values not surfaced). - `SCVec` / `SCMap` contract arguments are not yet encoded. -- `simulateTransaction`, `getEvents`, `getLedgerEntries`, `getFeeStats`, and TTL/footprint operations are not implemented. +- `simulateTransaction`, `getLedgerEntries`, `getFeeStats`, and TTL/footprint operations are not implemented. +- `getEvents` serves contract events only: `system` events are never emitted (the semantics have no source of them), and events whose topics/data have no staging representation (maps, errors, 256-bit ints) are dropped with a warning. `xdrFormat: "json"` is rejected. diff --git a/docs/server.md b/docs/server.md index 0af41c5..e89495d 100644 --- a/docs/server.md +++ b/docs/server.md @@ -15,6 +15,7 @@ class StellarRpcServer: receipts_dir: Path # io_dir / 'receipts' — receipt_.json per transaction traces_dir: Path # io_dir / 'traces' — trace_.jsonl per transaction requests_dir: Path # io_dir / 'requests' — request_.json archive + events_dir: Path # io_dir / 'events' — events_.json per ledger ``` The server is a plain `http.server.HTTPServer` (not pyk's `JsonRpcServer`). A `BaseHTTPRequestHandler` reads each POST body and calls `_handle`, which parses the JSON-RPC frame and delegates to `handle_rpc`. @@ -28,7 +29,7 @@ server = StellarRpcServer(io_dir=Path('out')) server.handle_rpc('sendTransaction', {'transaction': xdr}) ``` -For `sendTransaction` it builds the request envelope with `encoder.build_tx_request` and runs it with `interpreter.run`; for the read-only methods (`getHealth`, `getNetwork`, `getLatestLedger`, `getTransaction`, `traceTransaction`) it builds a small envelope and runs it. In every case the *content* of the response is produced by the semantics (`node.md`), not by Python — the one exception is the failure fallback (below). Each call is logged to stderr. +For `sendTransaction` it builds the request envelope with `encoder.build_tx_request` and runs it with `interpreter.run`; for the read-only methods (`getHealth`, `getNetwork`, `getLatestLedger`, `getTransaction`, `traceTransaction`, `getEvents`) it builds a small envelope and runs it. In every case the *content* of the response is produced by the semantics (`node.md`), not by Python — the one exception is the failure fallback (below). Each call is logged to stderr. --- @@ -51,7 +52,7 @@ At construction the server prepares the *io dir*, where `state.kore` lives at `i - **`state.kore` absent** — `interpreter.empty_config()` produces the initial idle K configuration (a blank-slate state with no accounts, contracts, or storage) and writes it; `metadata.json` is seeded with `{"latest_ledger": 0}`. - **`state.kore` present** — it is used as-is, and `metadata.json` is seeded only if missing. This lets you resume a previous session (ledger counter and stored receipts included) or start against a pre-built state. -In both cases the server creates the `receipts/`, `traces/`, and `requests/` directories if they do not already exist, because the K file-system hooks write into them but cannot create them. +In both cases the server creates the `receipts/`, `traces/`, `requests/`, and `events/` directories if they do not already exist, because the K file-system hooks write into them but cannot create them. Once the socket is bound, `serve` logs three lines to stderr: whether it is starting from a fresh state (an empty io-dir) or resuming an existing one (with the latest ledger), the io-dir path, and the listening address. Instruction tracing is always on, so every transaction the semantics run produces a trace. (Tracing only produces records for contract invocations.) @@ -143,6 +144,30 @@ All methods are answered by the K semantics and follow the [Stellar RPC specific `resultXdr` and `resultMetaXdr` are currently empty stubs. The receipt carries no trace — use `traceTransaction` with the same hash to fetch it. +### `getEvents` + +`getEvents` returns the contract events emitted in a ledger window — `startLedger` (inclusive) to `endLedger` (exclusive) — optionally narrowed by up to five filters (`type`, `contractIds`, `topics` with `*`/`**` wildcards) and paginated with `pagination.cursor` / `pagination.limit` (default 100). A cursor replaces `startLedger`/`endLedger` and resumes strictly after the last returned event. + +Events are captured during `sendTransaction`: the semantics intercept the `contract_event` host function and stage each event's topics and data, and after a successful run the server converts the staged records into one finished `events/events_.json` per ledger (`_finalize_events` — the base64 SCVal XDR, strkey, and TOID encodings are XDR work K cannot do). The K `getEvents` rules scan, filter, and paginate those files. Parameter validation lives in `_get_events_envelope`; `xdrFormat: "json"` is not supported and is rejected with `-32602`. + +**Response**: +```json +{ + "latestLedger": 4, + "events": [ + { + "type": "contract", "ledger": 4, "ledgerClosedAt": "2024-05-18T04:00:00Z", + "contractId": "C...", "id": "0000000017179873280-0000000000", + "inSuccessfulContractCall": true, "txHash": "<64-char hex>", + "topic": [""], "value": "" + } + ], + "cursor": "0000000021474836480-0000000000" +} +``` + +`system` events are never emitted (the semantics have no source of them), and events carrying values with no staging representation (maps, errors, 256-bit ints) are dropped with a warning rather than served with fabricated XDR. + --- ## Failure fallback diff --git a/src/komet_node/kdist/node.md b/src/komet_node/kdist/node.md index b8f1db2..cad9a85 100644 --- a/src/komet_node/kdist/node.md +++ b/src/komet_node/kdist/node.md @@ -54,7 +54,10 @@ module NODE | #enableTrace( String ) | #getTxResult( String, String, JSON, Int ) | #respondTrace( JSON, String ) + | #getEvents( JSON, Int ) + | #respondEvents( JSON, Int, JSONs ) | #respond( JSON, JSON ) + | #respondError( JSON, Int, String ) syntax Step ::= setLedgerSequence(Int) [symbol(setLedgerSequence)] // ---------------------------------------------------------------------- @@ -175,6 +178,22 @@ rule. `#respond(ID, RESULT)` writes the JSON-RPC envelope to `response.json`, re _ => 0 ``` +`#respondError(ID, CODE, MESSAGE)` is the error-shaped counterpart of `#respond`, for +requests that are recognised but cannot be answered (e.g. a ledger range outside the chain). + +```k + rule #respondError( ID, CODE, MESSAGE ) + => #writeFile( "response.json", JSON2String({ + "jsonrpc" : "2.0", + "id" : ID, + "error" : { "code" : CODE, "message" : MESSAGE } + })) + ~> #remove( "request.json" ) + ... + + _ => 0 +``` + ############################################################################### ## Read-only methods @@ -350,6 +369,221 @@ exists for that hash. requires notBool #fileExists( #traceFile( HASH ) ) ``` +## getEvents + +Contract events are captured during `sendTransaction` (see "Event capture" below) and +persisted by the Python server as one finished JSON array per ledger in +`events/events_.json` — each entry already in the spec's Event shape (base64 SCVal +XDR topics/value, strkey contract id, TOID-style id). `getEvents` scans the requested +ledger window, applies the request's filters, and paginates. + +The Python server validates the request parameters (types, filter/topic counts, cursor +format, `xdrFormat`) and guarantees the envelope carries: `startLedger` (Int or null), +`endLedger` (Int or null), `filters` (array), `cursor` (String or null, exclusive with +`startLedger`/`endLedger`), and `limit` (Int ≥ 1). Only the state-dependent check — the +requested window against the chain tip — is performed here. + +```k + rule #dispatchMethod( "getEvents", REQ ) + => #getEvents( REQ, #getInt( "latest_ledger", String2JSON( {#readFile("metadata.json")}:>String ) ) ) + ... + + + rule [getEvents-start-beyond-latest]: + #getEvents( REQ, LL ) + => #respondError( #getJSON( "id", REQ ), -32600, + "startLedger must be within the ledger range: 1 - " +String Int2String( LL ) ) + ... + + requires #getJSON( "cursor", REQ ) ==K null + andBool #getInt( "startLedger", REQ ) >Int LL + + rule [getEvents-scan]: + #getEvents( REQ, LL ) + => #respondEvents( REQ, LL, + #matchingEvents( #scanStart( REQ ), #scanEnd( REQ, LL ), #afterId( REQ ), #getJSON( "filters", REQ ) ) ) + ... + + [owise] + + rule #respondEvents( REQ, LL, MATCHED ) + => #respond( #getJSON( "id", REQ ), { + "latestLedger" : LL, + "events" : [ #takeJSONs( #getInt( "limit", REQ ), MATCHED ) ], + "cursor" : #pageCursor( MATCHED, #getInt( "limit", REQ ), #scanEnd( REQ, LL ) ) + }) + ... + +``` + +The scan window: `startLedger` is inclusive, `endLedger` exclusive, both capped at the +latest ledger. A pagination cursor replaces `startLedger` — scanning resumes at the +cursor's ledger (its TOID's high 32 bits) and only events with an id strictly greater than +the cursor are returned. Event ids are fixed-width zero-padded, so id order is plain string +order. + +```k + syntax Int ::= #scanStart( JSON ) [function] + // -------------------------------------------- + rule #scanStart( REQ ) => #cursorLedger( #getString( "cursor", REQ ) ) + requires #getJSON( "cursor", REQ ) =/=K null + rule #scanStart( REQ ) => #getInt( "startLedger", REQ ) [owise] + + syntax Int ::= #cursorLedger( String ) [function] + // ------------------------------------------------- + rule #cursorLedger( C ) => String2Int( substrString( C, 0, 19 ) ) >>Int 32 + + syntax Int ::= #scanEnd( JSON, Int ) [function] // exclusive end of the scan window + // ------------------------------------------------------------------------------------- + rule #scanEnd( REQ, LL ) => minInt( #getInt( "endLedger", REQ ), LL +Int 1 ) + requires #getJSON( "endLedger", REQ ) =/=K null + rule #scanEnd( _, LL ) => LL +Int 1 [owise] + + syntax String ::= #afterId( JSON ) [function] // only events with id > this are returned + // ------------------------------------------------------------------------------------------ + rule #afterId( REQ ) => #getString( "cursor", REQ ) + requires #getJSON( "cursor", REQ ) =/=K null + rule #afterId( _ ) => "" [owise] + + syntax String ::= #eventsFile( Int ) [function, symbol(eventsFile)] + // ------------------------------------------------------------------- + rule #eventsFile( L ) => "events/events_" +String Int2String( L ) +String ".json" + + syntax JSONs ::= #matchingEvents( Int, Int, String, JSON ) [function] + // --------------------------------------------------------------------- + rule #matchingEvents( L, END, _, _ ) => .JSONs requires L >=Int END + rule #matchingEvents( L, END, AFTER, FILTERS ) + => #concatJSONs( #filterEvents( #ledgerEvents( L ), AFTER, FILTERS ), + #matchingEvents( L +Int 1, END, AFTER, FILTERS ) ) + [owise] + + syntax JSONs ::= #ledgerEvents( Int ) [function] + // ------------------------------------------------ + rule #ledgerEvents( L ) => #arrayJSONs( String2JSON( {#readFile( #eventsFile( L ) )}:>String ) ) + requires #fileExists( #eventsFile( L ) ) + rule #ledgerEvents( _ ) => .JSONs [owise] + + syntax JSONs ::= #arrayJSONs( JSON ) [function] + // ----------------------------------------------- + rule #arrayJSONs( [ ES ] ) => ES + rule #arrayJSONs( _ ) => .JSONs [owise] + + syntax JSONs ::= #filterEvents( JSONs, String, JSON ) [function] + // ---------------------------------------------------------------- + rule #filterEvents( .JSONs, _, _ ) => .JSONs + rule #filterEvents( ( E, ES ), AFTER, FILTERS ) => ( E, #filterEvents( ES, AFTER, FILTERS ) ) + requires AFTER #filterEvents( ES, AFTER, FILTERS ) [owise] +``` + +An event passes if any filter matches (or the filter list is empty); a filter matches if +all of its present criteria — `type`, `contractIds`, `topics` — match. + +```k + syntax Bool ::= #matchesFilters( JSON, JSON ) [function] + // -------------------------------------------------------- + rule #matchesFilters( _, null ) => true + rule #matchesFilters( _, [ .JSONs ] ) => true + rule #matchesFilters( E, [ FS ] ) => #matchesAnyFilter( E, FS ) [owise] + + syntax Bool ::= #matchesAnyFilter( JSON, JSONs ) [function] + // ----------------------------------------------------------- + rule #matchesAnyFilter( _, .JSONs ) => false + rule #matchesAnyFilter( E, ( F, FS ) ) => #matchesFilter( E, F ) orBool #matchesAnyFilter( E, FS ) + + syntax Bool ::= #matchesFilter( JSON, JSON ) [function] + // ------------------------------------------------------- + rule #matchesFilter( E, F ) + => #matchesType( #getString( "type", E ), #getJSON( "type", F ) ) + andBool #matchesContractIds( #getString( "contractId", E ), #getJSON( "contractIds", F ) ) + andBool #matchesTopicFilters( #getJSON( "topic", E ), #getJSON( "topics", F ) ) + + syntax Bool ::= #matchesType( String, JSON ) [function] + // ------------------------------------------------------- + rule #matchesType( _, null ) => true + rule #matchesType( T, F:String ) => T ==String F + rule #matchesType( _, _ ) => false [owise] + + syntax Bool ::= #matchesContractIds( String, JSON ) [function] + // -------------------------------------------------------------- + rule #matchesContractIds( _, null ) => true + rule #matchesContractIds( _, [ .JSONs ] ) => true + rule #matchesContractIds( C, [ I0:JSON, IS ] ) => #containsString( C, ( I0, IS ) ) + rule #matchesContractIds( _, _ ) => false [owise] + + syntax Bool ::= #containsString( String, JSONs ) [function] + // ----------------------------------------------------------- + rule #containsString( _, .JSONs ) => false + rule #containsString( S, ( S2:String, SS ) ) => S ==String S2 orBool #containsString( S, SS ) + rule #containsString( S, ( _, SS ) ) => #containsString( S, SS ) [owise] +``` + +Topic filters: each filter is a list of segment matchers compared pairwise against the +event's topics — a base64 SCVal XDR string for an exact match, `"*"` for exactly one +segment, and a final `"**"` for zero or more remaining segments. + +```k + syntax Bool ::= #matchesTopicFilters( JSON, JSON ) [function] + // ------------------------------------------------------------- + rule #matchesTopicFilters( _, null ) => true + rule #matchesTopicFilters( _, [ .JSONs ] ) => true + rule #matchesTopicFilters( T, [ TF:JSON, TFS ] ) => #anyTopicFilter( T, ( TF, TFS ) ) + rule #matchesTopicFilters( _, _ ) => false [owise] + + syntax Bool ::= #anyTopicFilter( JSON, JSONs ) [function] + // --------------------------------------------------------- + rule #anyTopicFilter( _, .JSONs ) => false + rule #anyTopicFilter( T, ( TF, TFS ) ) => #matchesTopicFilter( T, TF ) orBool #anyTopicFilter( T, TFS ) + + syntax Bool ::= #matchesTopicFilter( JSON, JSON ) [function] + // ------------------------------------------------------------ + rule #matchesTopicFilter( [ TS ], [ MS ] ) => #matchSegments( TS, MS ) + rule #matchesTopicFilter( _, _ ) => false [owise] + + syntax Bool ::= #matchSegments( JSONs, JSONs ) [function] + // --------------------------------------------------------- + rule #matchSegments( _, ( "**", .JSONs ) ) => true + rule #matchSegments( .JSONs, .JSONs ) => true + rule #matchSegments( ( _:JSON, TS ), ( "*", MS ) ) => #matchSegments( TS, MS ) + rule #matchSegments( ( T:String, TS ), ( M:String, MS ) ) + => T ==String M andBool #matchSegments( TS, MS ) + requires M =/=String "*" andBool M =/=String "**" + rule #matchSegments( _, _ ) => false [owise] +``` + +Pagination: at most `limit` events are returned. The response cursor resumes the scan — the +id of the last returned event when the page filled up, otherwise a position just past the +scanned window (the TOID of the window's exclusive end ledger, which every later event id +exceeds). + +```k + syntax JSONs ::= #takeJSONs( Int, JSONs ) [function, total] + // ----------------------------------------------------------- + rule #takeJSONs( N, _ ) => .JSONs requires N <=Int 0 + rule #takeJSONs( N, .JSONs ) => .JSONs requires N >Int 0 + rule #takeJSONs( N, ( E, ES ) ) => ( E, #takeJSONs( N -Int 1, ES ) ) requires N >Int 0 + + syntax Int ::= #lengthJSONs( JSONs ) [function, total] + // ------------------------------------------------------ + rule #lengthJSONs( .JSONs ) => 0 + rule #lengthJSONs( ( _, ES ) ) => 1 +Int #lengthJSONs( ES ) + + syntax String ::= #lastEventId( JSONs ) [function] + // -------------------------------------------------- + rule #lastEventId( ( E, .JSONs ) ) => #getString( "id", E ) + rule #lastEventId( ( _, ES ) ) => #lastEventId( ES ) [owise] + + syntax String ::= #pageCursor( JSONs, Int, Int ) [function] + // ----------------------------------------------------------- + rule #pageCursor( MATCHED, LIMIT, ENDEXCL ) => #ledgerCursor( ENDEXCL ) + requires #lengthJSONs( MATCHED ) <=Int LIMIT + rule #pageCursor( MATCHED, LIMIT, _ ) => #lastEventId( #takeJSONs( LIMIT, MATCHED ) ) [owise] + + syntax String ::= #ledgerCursor( Int ) [function] + // ------------------------------------------------- + rule #ledgerCursor( L ) => #padLeft( Int2String( L < // clear the host cell before contract calls (_:HostCell => .HostStack ... ) +``` + +############################################################################### +# Event capture + +The upstream soroban semantics implement the `contract_event` host function ("x"/"1") as a +no-op that drops the topics and data. This rule shadows it (priority 40, ahead of the +upstream rule's default 50): it resolves the topics vector and the data value from the host +objects and appends one JSON record per event to `events_staged.jsonl` in the io-dir — +mirroring how the tracer appends to the trace file — before yielding the same `Void` result +as upstream. The Python server deletes the staging file before each transaction runs and, +after a successful `sendTransaction`, converts the staged records into the finished +`events/events_.json` that `getEvents` serves (base64 SCVal XDR and strkey ids are +XDR work that K cannot do). + +```k + rule [node-contract-event]: + hostCall ( "x" , "1" , [ i64 i64 .ValTypes ] -> [ i64 .ValTypes ] ) + => #stageEvent(HostVal(TOPICS), HostVal(DATA)) + ~> toSmall(Void) + ... + + + 0 |-> TOPICS + 1 |-> DATA + + [priority(40)] + + syntax InternalInstr ::= #stageEvent( HostVal, HostVal ) [symbol(stageEvent)] + // ------------------------------------------------------------------------------- + rule [stageEvent]: + #stageEvent(TOPICS, DATA) + => #appendFile( "events_staged.jsonl", + JSON2String({ + "contractId" : #bytesHex(ADDR), + "topics" : [ #topicsJSONs(HostVal2ScVal(TOPICS, OBJS, RELS)) ], + "data" : #scValJSON(HostVal2ScVal(DATA, OBJS, RELS)) + }) +String "\n" ) + ... + + Contract(ADDR) + OBJS + RELS +``` + +`#scValJSON` serialises a resolved `ScVal` in the same `{"type": ..., "value": ...}` scheme +that `#decodeArg` consumes (see `scval.py`), extended with `vec`, `string`, and `void`. +Values with no JSON representation here (maps, errors, 256-bit ints) become `null`; the +Python side skips such events with a warning rather than fabricating XDR for them. + +```k + syntax JSONs ::= #topicsJSONs( ScVal ) [function, total] + // -------------------------------------------------------- + rule #topicsJSONs( ScVec(L) ) => #scValsJSONs(L) + rule #topicsJSONs( _ ) => .JSONs [owise] + + syntax JSONs ::= #scValsJSONs( List ) [function, total] + // ------------------------------------------------------- + rule #scValsJSONs( .List ) => .JSONs + rule #scValsJSONs( ListItem(V:ScVal) L ) => ( #scValJSON(V), #scValsJSONs(L) ) + rule #scValsJSONs( ListItem(_) L ) => ( null, #scValsJSONs(L) ) [owise] + + syntax JSON ::= #scValJSON( ScVal ) [function, total] + // ----------------------------------------------------- + rule #scValJSON( SCBool(B) ) => { "type" : "bool" , "value" : B } + rule #scValJSON( Void ) => { "type" : "void" } + rule #scValJSON( U32(I) ) => { "type" : "u32" , "value" : I } + rule #scValJSON( I32(I) ) => { "type" : "i32" , "value" : I } + rule #scValJSON( U64(I) ) => { "type" : "u64" , "value" : I } + rule #scValJSON( I64(I) ) => { "type" : "i64" , "value" : I } + rule #scValJSON( U128(I) ) => { "type" : "u128" , "value" : I } + rule #scValJSON( I128(I) ) => { "type" : "i128" , "value" : I } + rule #scValJSON( Symbol(S) ) => { "type" : "symbol", "value" : S } + rule #scValJSON( ScString(S) ) => { "type" : "string", "value" : S } + rule #scValJSON( ScBytes(B) ) => { "type" : "bytes" , "value" : #bytesHex(B) } + rule #scValJSON( ScAddress(Account(B)) ) => { "type" : "address", "addrType" : "account" , "value" : #bytesHex(B) } + rule #scValJSON( ScAddress(Contract(B)) ) => { "type" : "address", "addrType" : "contract", "value" : #bytesHex(B) } + rule #scValJSON( ScVec(L) ) => { "type" : "vec" , "value" : [ #scValsJSONs(L) ] } + rule #scValJSON( _ ) => null [owise] + + syntax String ::= #bytesHex( Bytes ) [function, total] + // ------------------------------------------------------ + rule #bytesHex( B ) => #padLeft( Base2String( Bytes2Int(B, BE, Unsigned), 16 ), 2 *Int lengthBytes(B) ) + + syntax String ::= #padLeft( String, Int ) [function, total] + // ----------------------------------------------------------- + rule #padLeft( S, W ) => #padLeft( "0" +String S, W ) requires lengthString(S) S requires lengthString(S) >=Int W endmodule ``` diff --git a/src/komet_node/scval.py b/src/komet_node/scval.py index c379b60..bc86d02 100644 --- a/src/komet_node/scval.py +++ b/src/komet_node/scval.py @@ -20,14 +20,19 @@ SCSymbol, SCVec, ) +from stellar_sdk import xdr from stellar_sdk.xdr.sc_address_type import SCAddressType from stellar_sdk.xdr.sc_val_type import SCValType if TYPE_CHECKING: + from typing import Any + from komet.scval import SCValue from stellar_sdk.xdr.sc_address import SCAddress as XDRSCAddress from stellar_sdk.xdr.sc_val import SCVal +_U64_MASK = (1 << 64) - 1 + def scval_to_json(scval: SCVal) -> dict: """Encode a Stellar XDR SCVal as a JSON-serialisable dict for the node request envelope. @@ -79,6 +84,58 @@ def scval_to_json(scval: SCVal) -> dict: raise NotImplementedError(f'Unsupported SCVal type for JSON encoding: {scval.type}') +def scval_from_json(obj: Any) -> SCVal: + """Decode the ScVal JSON produced by the K semantics (``#scValJSON`` in ``node.md``) + into a Stellar XDR SCVal. + + This is the inverse of :func:`scval_to_json`, extended with the ``vec``, ``string``, + and ``void`` cases the event capture can stage. Values the semantics cannot represent + are staged as ``null`` and rejected here (``NotImplementedError``), as is any + unrecognised type tag. + """ + if not isinstance(obj, dict): + raise NotImplementedError(f'Unsupported staged SCVal encoding: {obj!r}') + value: Any = obj.get('value') + match obj.get('type'): + case 'bool': + return xdr.SCVal(type=SCValType.SCV_BOOL, b=bool(value)) + case 'void': + return xdr.SCVal(type=SCValType.SCV_VOID) + case 'u32': + return xdr.SCVal(type=SCValType.SCV_U32, u32=xdr.Uint32(value)) + case 'i32': + return xdr.SCVal(type=SCValType.SCV_I32, i32=xdr.Int32(value)) + case 'u64': + return xdr.SCVal(type=SCValType.SCV_U64, u64=xdr.Uint64(value)) + case 'i64': + return xdr.SCVal(type=SCValType.SCV_I64, i64=xdr.Int64(value)) + case 'u128': + u128 = xdr.UInt128Parts(hi=xdr.Uint64(value >> 64), lo=xdr.Uint64(value & _U64_MASK)) + return xdr.SCVal(type=SCValType.SCV_U128, u128=u128) + case 'i128': + i128 = xdr.Int128Parts(hi=xdr.Int64(value >> 64), lo=xdr.Uint64(value & _U64_MASK)) + return xdr.SCVal(type=SCValType.SCV_I128, i128=i128) + case 'symbol': + return xdr.SCVal(type=SCValType.SCV_SYMBOL, sym=xdr.SCSymbol(value.encode())) + case 'string': + return xdr.SCVal(type=SCValType.SCV_STRING, str=xdr.SCString(value.encode())) + case 'bytes': + return xdr.SCVal(type=SCValType.SCV_BYTES, bytes=xdr.SCBytes(bytes.fromhex(value))) + case 'address': + raw = bytes.fromhex(value) + if obj.get('addrType') == 'account': + key = xdr.PublicKey(xdr.PublicKeyType.PUBLIC_KEY_TYPE_ED25519, ed25519=xdr.Uint256(raw)) + address = xdr.SCAddress(SCAddressType.SC_ADDRESS_TYPE_ACCOUNT, account_id=xdr.AccountID(key)) + else: + contract = xdr.ContractID(xdr.Hash(raw)) + address = xdr.SCAddress(SCAddressType.SC_ADDRESS_TYPE_CONTRACT, contract_id=contract) + return xdr.SCVal(type=SCValType.SCV_ADDRESS, address=address) + case 'vec': + return xdr.SCVal(type=SCValType.SCV_VEC, vec=xdr.SCVec([scval_from_json(v) for v in value])) + case other: + raise NotImplementedError(f'Unsupported staged SCVal type: {other!r}') + + def sc_address_from_xdr(xdr: XDRSCAddress) -> SCAddress: """Convert an XDR SCAddress to a Komet SCAddress.""" match xdr.type: diff --git a/src/komet_node/server.py b/src/komet_node/server.py index 6399575..132bae9 100644 --- a/src/komet_node/server.py +++ b/src/komet_node/server.py @@ -2,17 +2,20 @@ import json import logging +import re import sys import tempfile import time import traceback +from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path from typing import TYPE_CHECKING, Any, Final -from stellar_sdk import Network +from stellar_sdk import Network, StrKey from komet_node.interpreter import NodeInterpreter +from komet_node.scval import scval_from_json from komet_node.transaction import TransactionEncoder if TYPE_CHECKING: @@ -24,6 +27,23 @@ # trace stored on a previously executed transaction's receipt (see _read_only_envelope). _TX_METHODS: Final = ('sendTransaction',) +# Where the K semantics stage the contract events of the currently executing transaction +# (one JSON record per line, appended by the `contract_event` interception in node.md). +_EVENTS_STAGING: Final = 'events_staged.jsonl' + +# getEvents cursors and event ids are TOID-style: a 19-digit zero-padded TOID, a hyphen, +# and a 10-digit zero-padded event index (SEP-35). +_EVENT_CURSOR_RE: Final = re.compile(r'\d{19}-\d{10}') + +# getEvents limits from the spec: at most 5 filters, 5 topic filters each, 1-4 segment +# matchers per topic filter (a trailing '**' does not count), and a page limit of 1-10000 +# (default 100). +_MAX_EVENT_FILTERS: Final = 5 +_MAX_TOPIC_FILTERS: Final = 5 +_MAX_TOPIC_SEGMENTS: Final = 4 +_DEFAULT_EVENTS_LIMIT: Final = 100 +_MAX_EVENTS_LIMIT: Final = 10000 + _log = logging.getLogger('komet_node') @@ -86,7 +106,10 @@ def __init__( self.receipts_dir = self.io_dir / 'receipts' self.traces_dir = self.io_dir / 'traces' self.requests_dir = self.io_dir / 'requests' - for directory in (self.receipts_dir, self.traces_dir, self.requests_dir): + # events/ holds one finished JSON array per ledger (events_.json), written by + # _finalize_events below and read back by the K getEvents rules. + self.events_dir = self.io_dir / 'events' + for directory in (self.receipts_dir, self.traces_dir, self.requests_dir, self.events_dir): directory.mkdir(exist_ok=True) # Continue the request archive numbering past anything a previous run left behind, so # resuming an io-dir never overwrites its earlier request files. @@ -195,9 +218,14 @@ def handle_rpc(self, method: str | None, params: dict[str, Any], request_id: Any # the client-facing message neutral rather than leaking internal exceptions. traceback.print_exc() return _error_str(request_id, -32602, 'Invalid params: could not process transaction') + # Stale staged events (e.g. from a previously failed transaction) must not leak + # into this transaction's event records. + (self.io_dir / _EVENTS_STAGING).unlink(missing_ok=True) response = self.interpreter.run(self.state_file, self.io_dir, envelope, program_steps) if response is None: + (self.io_dir / _EVENTS_STAGING).unlink(missing_ok=True) return json.dumps(self._failure_response(request_id, envelope, now)) + self._finalize_events(envelope['txHash'], now) return response read_only_envelope = self._read_only_envelope(method, params, request_id, now) @@ -230,8 +258,111 @@ def _read_only_envelope( if not isinstance(tx_hash, str): return _error_str(request_id, -32602, "Invalid params: 'hash' (string) is required") return {**base, 'hash': tx_hash} + if method == 'getEvents': + return self._get_events_envelope(base, params, request_id) return None + def _get_events_envelope( + self, base: dict[str, Any], params: dict[str, Any], request_id: Any + ) -> dict[str, Any] | str: + """Validate the getEvents params and build the request envelope. + + Everything structural is checked here so the K rules (node.md) only handle + well-typed envelopes; the one state-dependent check (the window against the chain + tip) lives in K. Returns a pre-formatted JSON-RPC error string on invalid params. + """ + xdr_format = params.get('xdrFormat') or 'base64' + if xdr_format == 'json': + return _error_str(request_id, -32602, "Invalid params: xdrFormat 'json' is not supported, use 'base64'") + if xdr_format != 'base64': + return _error_str(request_id, -32602, "Invalid params: xdrFormat must be 'base64' or 'json'") + + pagination = params.get('pagination') or {} + if not isinstance(pagination, dict): + return _error_str(request_id, -32602, "Invalid params: 'pagination' must be an object") + cursor = pagination.get('cursor') + if cursor is not None and not (isinstance(cursor, str) and _EVENT_CURSOR_RE.fullmatch(cursor)): + return _error_str(request_id, -32602, "Invalid params: 'cursor' is malformed") + limit = pagination.get('limit') + if limit is None: + limit = _DEFAULT_EVENTS_LIMIT + if isinstance(limit, float) and limit.is_integer(): + limit = int(limit) + if not _is_int(limit) or not 1 <= limit <= _MAX_EVENTS_LIMIT: + return _error_str(request_id, -32602, f'Invalid params: limit must be between 1 and {_MAX_EVENTS_LIMIT}') + + start_ledger = params.get('startLedger') + end_ledger = params.get('endLedger') + if cursor is not None and (start_ledger is not None or end_ledger is not None): + return _error_str(request_id, -32600, 'startLedger and endLedger must be omitted when a cursor is set') + if cursor is None and start_ledger is None: + return _error_str(request_id, -32600, 'startLedger or a pagination cursor is required') + for name, value in (('startLedger', start_ledger), ('endLedger', end_ledger)): + if value is not None and not _is_int(value): + return _error_str(request_id, -32602, f'Invalid params: {name} must be a ledger sequence number') + if start_ledger is not None and start_ledger < 1: + return _error_str(request_id, -32600, 'startLedger must be positive') + + filters = params.get('filters') or [] + error = _validate_event_filters(filters) + if error is not None: + return _error_str(request_id, -32602, f'Invalid params: {error}') + + return { + **base, + 'startLedger': start_ledger, + 'endLedger': end_ledger, + 'filters': filters, + 'cursor': cursor, + 'limit': limit, + } + + def _finalize_events(self, tx_hash: str, now: str) -> None: + """Convert the events staged by the K run into ``events/events_.json``. + + The semantics stage one JSON record per contract event (K-side ScVal JSON, hex + contract id — see "Event capture" in node.md); this turns each into the spec's + Event shape — base64 SCVal XDR topics/value, strkey contract id, TOID-style id — + which K cannot build itself. Runs after every successful sendTransaction; with + nothing staged it only removes the staging file. + """ + staging = self.io_dir / _EVENTS_STAGING + if not staging.exists(): + return + lines = [line for line in staging.read_text().splitlines() if line.strip()] + staging.unlink() + if not lines: + return + metadata = json.loads((self.io_dir / 'metadata.json').read_text()) + ledger = metadata['latest_ledger'] + closed_at = datetime.fromtimestamp(int(now), tz=timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') + # One transaction per ledger: TOID with application order 1, operation index 0. + toid = (ledger << 32) | (1 << 12) + events = [] + for index, line in enumerate(lines): + staged = json.loads(line) + try: + topic = [scval_from_json(t).to_xdr() for t in staged['topics']] + value = scval_from_json(staged['data']).to_xdr() + except NotImplementedError as err: + _log.warning('dropping contract event with unsupported value: %s', err) + continue + events.append( + { + 'type': 'contract', + 'ledger': ledger, + 'ledgerClosedAt': closed_at, + 'contractId': StrKey.encode_contract(bytes.fromhex(staged['contractId'])), + 'id': f'{toid:019d}-{index:010d}', + 'inSuccessfulContractCall': True, + 'txHash': tx_hash, + 'topic': topic, + 'value': value, + } + ) + if events: + (self.events_dir / f'events_{ledger}.json').write_text(json.dumps(events)) + def _archive_request(self, method: str | None, params: dict[str, Any], request_id: Any) -> None: """Write each incoming JSON-RPC call to its own ``requests/request_.json`` file. @@ -276,6 +407,47 @@ def _failure_response(self, rpc_id: Any, envelope: dict[str, Any], now: str) -> return {'jsonrpc': '2.0', 'id': rpc_id, 'result': result} +def _is_int(value: Any) -> bool: + """True for a JSON integer (bool is an int subclass in Python, so exclude it).""" + return isinstance(value, int) and not isinstance(value, bool) + + +def _validate_event_filters(filters: Any) -> str | None: + """Check a getEvents ``filters`` param against the EventFilters schema. + + Returns a human-readable problem description, or ``None`` if the filters are valid. + """ + if not isinstance(filters, list): + return "'filters' must be an array" + if len(filters) > _MAX_EVENT_FILTERS: + return f'maximum {_MAX_EVENT_FILTERS} filters per request' + for event_filter in filters: + if not isinstance(event_filter, dict): + return 'each filter must be an object' + event_type = event_filter.get('type') + if event_type is not None and event_type not in ('contract', 'system'): + return "filter 'type' must be 'contract' or 'system'" + contract_ids = event_filter.get('contractIds') + if contract_ids is not None: + if not isinstance(contract_ids, list): + return "'contractIds' must be an array" + if not all(isinstance(c, str) and StrKey.is_valid_contract(c) for c in contract_ids): + return "each entry of 'contractIds' must be a contract address" + topics = event_filter.get('topics') + if topics is not None: + if not isinstance(topics, list) or len(topics) > _MAX_TOPIC_FILTERS: + return f"'topics' must be an array of at most {_MAX_TOPIC_FILTERS} topic filters" + for topic_filter in topics: + if not isinstance(topic_filter, list) or not all(isinstance(s, str) for s in topic_filter): + return 'each topic filter must be an array of segment matchers' + # A trailing '**' matches any remaining segments and does not count + # towards the 1-4 segment matcher limit. + segments = topic_filter[:-1] if topic_filter and topic_filter[-1] == '**' else topic_filter + if not 1 <= len(segments) <= _MAX_TOPIC_SEGMENTS or '**' in segments: + return f'each topic filter must hold 1 to {_MAX_TOPIC_SEGMENTS} segment matchers' + return None + + def _next_request_index(requests_dir: Path) -> int: """Return the next free index for ``requests/request_.json``. diff --git a/src/tests/integration/data/wasm/events.wat b/src/tests/integration/data/wasm/events.wat new file mode 100644 index 0000000..5207676 --- /dev/null +++ b/src/tests/integration/data/wasm/events.wat @@ -0,0 +1,41 @@ +(module + (type (;0;) (func)) + (type (;1;) (func (result i64))) + (type (;2;) (func (param i64 i64) (result i64))) + + ;; Soroban host imports (the semantics resolve these to hostCall): + ;; v._ vec_new() -> VecObject + ;; v.6 vec_push_back(vec, val) -> VecObject + ;; x.1 contract_event(topics, data) -> Void + (import "v" "_" (func (;0;) (type 1))) + (import "v" "6" (func (;1;) (type 2))) + (import "x" "1" (func (;2;) (type 2))) + + ;; emit: publish one contract event with topics [Symbol("transfer")] and + ;; data U32(42), then return Void. + ;; + ;; Small-value encodings per CAP-46-01 (tag in the low 8 bits): + ;; Symbol("transfer") = (encode6bit("transfer") << 8) | 14 = 65154533130155790 + ;; U32(42) = (42 << 32) | 4 = 180388626436 + ;; Void = 2 + (func (;3;) (type 1) (result i64) + call 0 ;; vec_new() -> empty topics vec + i64.const 65154533130155790 ;; Symbol("transfer") + call 1 ;; vec_push_back(vec, symbol) + i64.const 180388626436 ;; U32(42) + call 2 ;; contract_event(topics, data) + drop + i64.const 2) ;; Void + + ;; _ (Soroban ABI stub) + (func (;4;) (type 0)) + + (memory (;0;) 16) + (global (;0;) (mut i32) (i32.const 1048576)) + (global (;1;) i32 (i32.const 1048576)) + (global (;2;) i32 (i32.const 1048576)) + + (export "memory" (memory 0)) + (export "emit" (func 3)) + (export "_" (func 4)) +) diff --git a/src/tests/integration/test_get_events.py b/src/tests/integration/test_get_events.py new file mode 100644 index 0000000..3b562e3 --- /dev/null +++ b/src/tests/integration/test_get_events.py @@ -0,0 +1,332 @@ +"""Integration tests for the ``getEvents`` JSON-RPC method. + +The expected behavior follows the official stellar-rpc OpenRPC spec (getEvents.json and the +Event / EventFilters / Cursor schemas): a request selects a ledger range with ``startLedger`` +(inclusive) / ``endLedger`` (exclusive) or resumes from a pagination ``cursor``, optionally +narrowed by up to five filters (event type, contract ids, topic matchers). The result carries +``events`` (array), ``latestLedger`` (JSON number) and ``cursor`` (string). + +The event-emitting contract lives in ``data/wasm/events.wat``: its ``emit`` function publishes +one contract event with topics ``[Symbol("transfer")]`` and data ``U32(42)`` via the +``contract_event`` host function (module "x", name "1"). +""" + +from __future__ import annotations + +import json +import re +import socket +import subprocess +import threading +import time +import urllib.request +from datetime import datetime +from pathlib import Path +from typing import TYPE_CHECKING, Any + +import pytest +from stellar_sdk import Account, Keypair, Network, StrKey, TransactionBuilder, xdr +from stellar_sdk.utils import sha256 +from stellar_sdk.xdr.sc_val_type import SCValType + +from komet_node.server import StellarRpcServer + +if TYPE_CHECKING: + from collections.abc import Iterator + +EVENTS_CONTRACT_WAT = (Path(__file__).parent / 'data' / 'wasm' / 'events.wat').resolve(strict=True) + +# Base64 SCVal XDR of the topic and data emitted by events.wat's `emit` function. +TRANSFER_TOPIC_XDR = xdr.SCVal(type=SCValType.SCV_SYMBOL, sym=xdr.SCSymbol(sc_symbol=b'transfer')).to_xdr() +MINT_TOPIC_XDR = xdr.SCVal(type=SCValType.SCV_SYMBOL, sym=xdr.SCSymbol(sc_symbol=b'mint')).to_xdr() +U32_42_XDR = xdr.SCVal(type=SCValType.SCV_U32, u32=xdr.Uint32(42)).to_xdr() + +# TOID-style event id: 19-digit zero-padded TOID, hyphen, 10-digit zero-padded event index +# (see the EventId schema / SEP-35). +EVENT_ID_RE = re.compile(r'\d{19}-\d{10}') + + +# --------------------------------------------------------------------------- +# Helpers (mirrored from test_server.py so this module stays self-contained) +# --------------------------------------------------------------------------- + + +def wat_to_wasm(wat_path: Path) -> bytes: + proc_res = subprocess.run(['wat2wasm', str(wat_path), '--output=/dev/stdout'], check=True, capture_output=True) + return proc_res.stdout + + +def _find_free_port() -> int: + with socket.socket() as s: + s.bind(('', 0)) + return s.getsockname()[1] + + +def _wait_for_server(host: str, port: int, timeout: float = 10.0) -> None: + deadline = time.time() + timeout + while time.time() < deadline: + try: + with socket.create_connection((host, port), timeout=0.1): + return + except OSError: + time.sleep(0.05) + raise TimeoutError(f'Server did not start on {host}:{port}') + + +def _rpc(port: int, method: str, params: dict[str, Any]) -> dict[str, Any]: + body = json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': method, 'params': params}).encode() + req = urllib.request.Request( + f'http://localhost:{port}', + data=body, + headers={'Content-Type': 'application/json'}, + ) + with urllib.request.urlopen(req) as resp: + return json.loads(resp.read()) + + +@pytest.fixture +def server(tmp_path: Path) -> Iterator[StellarRpcServer]: + port = _find_free_port() + srv = StellarRpcServer( + host='localhost', + port=port, + io_dir=tmp_path, + network_passphrase=Network.TESTNET_NETWORK_PASSPHRASE, + ) + thread = threading.Thread(target=srv.serve, daemon=True) + thread.start() + _wait_for_server('localhost', port) + yield srv + srv.shutdown() + + +def _send(server: StellarRpcServer, keypair: Keypair, tb: TransactionBuilder) -> str: + """Sign, submit, and confirm a transaction; return its hash.""" + env = tb.set_timeout(30).build() + env.sign(keypair) + res = _rpc(server.port(), 'sendTransaction', {'transaction': env.to_xdr()}) + assert res['result']['status'] == 'PENDING' + tx_hash = res['result']['hash'] + get_res = _rpc(server.port(), 'getTransaction', {'hash': tx_hash})['result'] + assert get_res['status'] == 'SUCCESS', f'Transaction failed: {get_res}' + return tx_hash + + +def _deploy_events_contract(server: StellarRpcServer) -> tuple[Keypair, Account, str]: + """Create an account, upload events.wat, and deploy it (ledgers 1-3). + + Returns the funding keypair, its (mutated) sequence-tracking account, and the C-strkey + address of the deployed contract. + """ + keypair = Keypair.random() + account = Account(keypair.public_key, sequence=0) + + def builder() -> TransactionBuilder: + return TransactionBuilder(account, Network.TESTNET_NETWORK_PASSPHRASE) + + _send(server, keypair, builder().append_create_account_op(keypair.public_key, '1000')) + + wasm_bytecode = wat_to_wasm(EVENTS_CONTRACT_WAT) + _send(server, keypair, builder().append_upload_contract_wasm_op(wasm_bytecode)) + + wasm_hash = sha256(wasm_bytecode) + salt = b'\x00' * 32 + _send(server, keypair, builder().append_create_contract_op(wasm_hash, keypair.public_key, None, salt)) + + contract_address = server.encoder.contract_address_from_deployer_address(keypair.public_key, salt) + return keypair, account, contract_address + + +def _emit_event(server: StellarRpcServer, keypair: Keypair, account: Account, contract_address: str) -> str: + """Invoke the contract's `emit` function; return the transaction hash.""" + tb = TransactionBuilder(account, Network.TESTNET_NETWORK_PASSPHRASE) + return _send(server, keypair, tb.append_invoke_contract_function_op(contract_address, 'emit', [])) + + +def _is_json_number(value: Any) -> bool: + """True for a JSON number decoded to int (bool is an int subclass, so exclude it).""" + return isinstance(value, int) and not isinstance(value, bool) + + +def _assert_request_error(response: dict[str, Any]) -> None: + """The call must be rejected with a JSON-RPC request/params error. + + The OpenRPC spec documents which requests are invalid but not the numeric code; real + stellar-rpc uses -32600 (Invalid Request) for getEvents request validation, and -32602 + (Invalid params) is the JSON-RPC code for bad params. Accept either, but nothing else. + """ + assert 'error' in response, f'expected an error response, got: {response}' + assert response['error']['code'] in (-32600, -32602), response['error'] + + +# --------------------------------------------------------------------------- +# Request validation +# --------------------------------------------------------------------------- + + +def test_get_events_requires_start_ledger_or_cursor(server: StellarRpcServer) -> None: + """Neither startLedger nor a pagination cursor: the request is invalid (per StartLedger).""" + _assert_request_error(_rpc(server.port(), 'getEvents', {})) + + +def test_get_events_rejects_start_ledger_beyond_latest(server: StellarRpcServer) -> None: + """startLedger greater than the latest ledger seen by the node is an error (per StartLedger).""" + _assert_request_error(_rpc(server.port(), 'getEvents', {'startLedger': 999999})) + + +def test_get_events_rejects_cursor_combined_with_start_ledger(server: StellarRpcServer) -> None: + """If a cursor is included, startLedger must be omitted (per StartLedger/EndLedger).""" + response = _rpc( + server.port(), + 'getEvents', + {'startLedger': 1, 'pagination': {'cursor': '0000000004294967296-0000000000'}}, + ) + _assert_request_error(response) + + +def test_get_events_rejects_more_than_five_filters(server: StellarRpcServer) -> None: + """EventFilters allows at most 5 filters per request.""" + filters = [{'type': 'contract'}] * 6 + _assert_request_error(_rpc(server.port(), 'getEvents', {'startLedger': 1, 'filters': filters})) + + +def test_get_events_rejects_topic_filter_with_five_segments(server: StellarRpcServer) -> None: + """A TopicFilter holds 1 to 4 SegmentMatchers.""" + filters = [{'topics': [[TRANSFER_TOPIC_XDR] * 5]}] + _assert_request_error(_rpc(server.port(), 'getEvents', {'startLedger': 1, 'filters': filters})) + + +def test_get_events_rejects_invalid_xdr_format(server: StellarRpcServer) -> None: + """xdrFormat only admits 'base64' and 'json'; anything else is invalid params. + + komet-node does not implement the JSON XDR representation, so 'json' is also rejected + with -32602 (documented limitation, consistent with the other methods). + """ + response = _rpc(server.port(), 'getEvents', {'startLedger': 1, 'xdrFormat': 'bogus'}) + assert 'error' in response, f'expected an error response, got: {response}' + assert response['error']['code'] == -32602 + + response = _rpc(server.port(), 'getEvents', {'startLedger': 1, 'xdrFormat': 'json'}) + assert 'error' in response, f'expected an error response, got: {response}' + assert response['error']['code'] == -32602 + + +# --------------------------------------------------------------------------- +# Result shape +# --------------------------------------------------------------------------- + + +def test_get_events_no_events_returns_empty_array(server: StellarRpcServer) -> None: + """A range that contains no contract events yields an empty events array, not null.""" + keypair = Keypair.random() + account = Account(keypair.public_key, sequence=0) + tb = TransactionBuilder(account, Network.TESTNET_NETWORK_PASSPHRASE) + _send(server, keypair, tb.append_create_account_op(keypair.public_key, '1000')) + + result = _rpc(server.port(), 'getEvents', {'startLedger': 1})['result'] + assert result['events'] == [] + assert _is_json_number(result['latestLedger']) + assert result['latestLedger'] == 1 + # Real stellar-rpc always populates the cursor (the position to resume scanning from). + assert isinstance(result['cursor'], str) + + +def test_get_events_returns_emitted_contract_event(server: StellarRpcServer) -> None: + """An event published via contract_event is returned with the exact spec shape.""" + keypair, account, contract_address = _deploy_events_contract(server) + tx_hash = _emit_event(server, keypair, account, contract_address) # ledger 4 + + result = _rpc(server.port(), 'getEvents', {'startLedger': 1})['result'] + assert _is_json_number(result['latestLedger']) + assert result['latestLedger'] == 4 + assert isinstance(result['cursor'], str) + + assert isinstance(result['events'], list) + assert len(result['events']) == 1, f'expected exactly one event: {result["events"]}' + event = result['events'][0] + + assert event['type'] == 'contract' + assert _is_json_number(event['ledger']) + assert event['ledger'] == 4 + # ledgerClosedAt is an ISO-8601 timestamp string. + assert isinstance(event['ledgerClosedAt'], str) + datetime.fromisoformat(event['ledgerClosedAt'].replace('Z', '+00:00')) + assert event['contractId'] == contract_address + assert StrKey.is_valid_contract(event['contractId']) + # The id is TOID-style: 19-digit TOID, hyphen, 10-digit event index; the TOID's high + # 32 bits are the ledger sequence. + assert EVENT_ID_RE.fullmatch(event['id']), event['id'] + assert int(event['id'].split('-')[0]) >> 32 == 4 + assert event['txHash'] == tx_hash + # Topics and value are base64-encoded SCVal XDR. + assert event['topic'] == [TRANSFER_TOPIC_XDR] + assert event['value'] == U32_42_XDR + # Deprecated but still emitted by stellar-rpc v22; if present it must be a true bool + # (the transaction succeeded). + if 'inSuccessfulContractCall' in event: + assert event['inSuccessfulContractCall'] is True + + # endLedger is exclusive: a window that ends at the event's ledger does not include it. + result = _rpc(server.port(), 'getEvents', {'startLedger': 1, 'endLedger': 4})['result'] + assert result['events'] == [] + + # xdrFormat 'base64' is the default and must be accepted explicitly as well. + result = _rpc(server.port(), 'getEvents', {'startLedger': 1, 'xdrFormat': 'base64'})['result'] + assert len(result['events']) == 1 + + +# --------------------------------------------------------------------------- +# Filters and pagination +# --------------------------------------------------------------------------- + + +def test_get_events_filtering_and_pagination(server: StellarRpcServer) -> None: + """Filters narrow by contract id, type, and topics; pagination resumes from the cursor.""" + keypair, account, contract_address = _deploy_events_contract(server) + _emit_event(server, keypair, account, contract_address) # ledger 4 + _emit_event(server, keypair, account, contract_address) # ledger 5 + + def get_events(params: dict[str, Any]) -> dict[str, Any]: + response = _rpc(server.port(), 'getEvents', params) + assert 'result' in response, f'expected a result, got: {response}' + return response['result'] + + # No filters: both events, in order. + result = get_events({'startLedger': 1}) + assert [e['ledger'] for e in result['events']] == [4, 5] + + # A matching contractIds filter keeps the events. + result = get_events({'startLedger': 1, 'filters': [{'contractIds': [contract_address]}]}) + assert len(result['events']) == 2 + + # A non-matching contract id filters everything out. + other_contract = StrKey.encode_contract(b'\x11' * 32) + result = get_events({'startLedger': 1, 'filters': [{'contractIds': [other_contract]}]}) + assert result['events'] == [] + + # Type filter: both events are contract events; no system events were emitted. + result = get_events({'startLedger': 1, 'filters': [{'type': 'contract'}]}) + assert len(result['events']) == 2 + result = get_events({'startLedger': 1, 'filters': [{'type': 'system'}]}) + assert result['events'] == [] + + # Topic filters: exact segment match, single-segment wildcard, and a non-matching topic. + result = get_events({'startLedger': 1, 'filters': [{'topics': [[TRANSFER_TOPIC_XDR]]}]}) + assert len(result['events']) == 2 + result = get_events({'startLedger': 1, 'filters': [{'topics': [['*']]}]}) + assert len(result['events']) == 2 + result = get_events({'startLedger': 1, 'filters': [{'topics': [[MINT_TOPIC_XDR]]}]}) + assert result['events'] == [] + + # Pagination: limit the first page to one event, then resume from the cursor (without + # startLedger, per the StartLedger descriptor) to fetch the second. + first_page = get_events({'startLedger': 1, 'pagination': {'limit': 1}}) + assert len(first_page['events']) == 1 + assert first_page['events'][0]['ledger'] == 4 + cursor = first_page['cursor'] + assert isinstance(cursor, str) and cursor != '' + + second_page = get_events({'pagination': {'cursor': cursor, 'limit': 1}}) + assert len(second_page['events']) == 1 + assert second_page['events'][0]['ledger'] == 5 + assert second_page['events'][0]['id'] != first_page['events'][0]['id']