diff --git a/docs/release-notes/change-log.md b/docs/release-notes/change-log.md index 23fff90b4..75c103eee 100644 --- a/docs/release-notes/change-log.md +++ b/docs/release-notes/change-log.md @@ -11,6 +11,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), ## Unreleased +### Added + +- WASM: new `context` host module giving modules an intrinsic they can call at any point during execution: `context::clock(output_ptr)` writes the block clock as an encoded `sf.substreams.v1.Clock`. It writes a `{ptr, len}` pair at `output_ptr`, the same convention the `state` getters use, and is available on the `wasmtime` and `wazero` runtimes (not on the JavaScript/v8 one). Until now the clock was only reachable by declaring `source: sf.substreams.v1.Clock` as a module input. Ergonomic Rust bindings will follow in `substreams-rs`; until then a module declares the import itself with `#[link(wasm_import_module = "context")]`. `context` joins `env`, `state` and `logger` as a namespace WASM extensions cannot register into. + ### Fixed - Server: `substreams-tier1` now restarts when its block hub can no longer link incoming live blocks, instead of hanging every request at a frozen head indefinitely. A live-source gap whose one-block files were already merged away can never be linked, and the head-block metrics keep tracking the live source, so the process looked healthy throughout. diff --git a/test/testdata/complex_substreams/complex-substreams-v0.1.0.spkg b/test/testdata/complex_substreams/complex-substreams-v0.1.0.spkg index c974ea7e5..be40baf99 100644 Binary files a/test/testdata/complex_substreams/complex-substreams-v0.1.0.spkg and b/test/testdata/complex_substreams/complex-substreams-v0.1.0.spkg differ diff --git a/test/testdata/complex_substreams/src/lib.rs b/test/testdata/complex_substreams/src/lib.rs index 71b7c3bc9..4204343b1 100644 --- a/test/testdata/complex_substreams/src/lib.rs +++ b/test/testdata/complex_substreams/src/lib.rs @@ -326,3 +326,36 @@ fn assert_set_sum_store_deltas_0(block: test::Block, set_sum_store: store::Delta + +/// Raw binding to the `context` host module. It lives here rather than coming from +/// `substreams` because the crate does not expose it yet. +mod context_intrinsics { + use prost::Message; + + #[link(wasm_import_module = "context")] + extern "C" { + fn clock(output_ptr: *mut u8); + } + + /// Reads the `{ptr, len}` pair the host writes at `output_ptr` and takes ownership of + /// the payload it points at, which the host allocated through our own `alloc`. + unsafe fn read_output(intrinsic: unsafe extern "C" fn(*mut u8)) -> Vec { + let mut output: [u32; 2] = [0, 0]; + intrinsic(output.as_mut_ptr() as *mut u8); + + let (ptr, len) = (output[0] as *mut u8, output[1] as usize); + Vec::from_raw_parts(ptr, len, len) + } + + pub fn get_clock() -> crate::pb::sf::substreams::v1::Clock { + let data = unsafe { read_output(clock) }; + crate::pb::sf::substreams::v1::Clock::decode(data.as_slice()).expect("clock is not a valid Clock message") + } +} + +#[substreams::handlers::map] +fn assert_context_intrinsics_0(clock: pb::sf::substreams::v1::Clock) -> Result { + assert_eq!(context_intrinsics::get_clock(), clock); + + Ok(test::Boolean { result: true }) +} diff --git a/test/testdata/complex_substreams/src/pb/mod.rs b/test/testdata/complex_substreams/src/pb/mod.rs index 527e59ffc..deedff8a6 100644 --- a/test/testdata/complex_substreams/src/pb/mod.rs +++ b/test/testdata/complex_substreams/src/pb/mod.rs @@ -1,11 +1,4 @@ // @generated -pub mod google { - // @@protoc_insertion_point(attribute:google.protobuf) - pub mod protobuf { - include!("google.protobuf.rs"); - // @@protoc_insertion_point(google.protobuf) - } -} pub mod sf { // @@protoc_insertion_point(attribute:sf.substreams) pub mod substreams { @@ -18,22 +11,6 @@ pub mod sf { // @@protoc_insertion_point(sf.substreams.index.v1) } } - pub mod rpc { - // @@protoc_insertion_point(attribute:sf.substreams.rpc.v2) - pub mod v2 { - include!("sf.substreams.rpc.v2.rs"); - // @@protoc_insertion_point(sf.substreams.rpc.v2) - } - } - pub mod sink { - pub mod service { - // @@protoc_insertion_point(attribute:sf.substreams.sink.service.v1) - pub mod v1 { - include!("sf.substreams.sink.service.v1.rs"); - // @@protoc_insertion_point(sf.substreams.sink.service.v1) - } - } - } // @@protoc_insertion_point(attribute:sf.substreams.v1) pub mod v1 { include!("sf.substreams.v1.rs"); diff --git a/test/testdata/complex_substreams/substreams.yaml b/test/testdata/complex_substreams/substreams.yaml index b0d86a332..52acd3091 100644 --- a/test/testdata/complex_substreams/substreams.yaml +++ b/test/testdata/complex_substreams/substreams.yaml @@ -35,6 +35,14 @@ modules: output: type: proto:sf.substreams.v1.test.Boolean + - name: assert_context_intrinsics_0 + kind: map + initialBlock: 0 + inputs: + - source: sf.substreams.v1.Clock + output: + type: proto:sf.substreams.v1.test.Boolean + - name: map_hybrid_input_clock_70 kind: map initialBlock: 70 diff --git a/test/tier2_integration_test.go b/test/tier2_integration_test.go index 8fa9de478..70627641c 100644 --- a/test/tier2_integration_test.go +++ b/test/tier2_integration_test.go @@ -56,6 +56,7 @@ func TestTier2Call(t *testing.T) { mapHybridInputClock70 := hex.EncodeToString([]byte("map_hybrid_input_clock_70")) mapHybridInputBlock70 := hex.EncodeToString([]byte("map_hybrid_input_block_70")) setSumStoreInit0 := hex.EncodeToString([]byte("set_sum_store_init_0")) + assertContextIntrinsics0 := hex.EncodeToString([]byte("assert_context_intrinsics_0")) randomIndicesRange := roaring64.New() randomIndicesRange.AddInt(70) @@ -357,6 +358,37 @@ func TestTier2Call(t *testing.T) { mapOutputFileToCheck: mapUsingIndexInit70 + "/outputs/0000000070-0000000080.output", expectedSkippedBlocks: map[uint64]struct{}{75: {}, 77: {}, 78: {}, 79: {}, 80: {}}, // faked with the randomIndicesRange above }, + // The module asserts, inside the WASM, that `context::clock` matches the Clock it + // received as a module input. A mismatch panics in the module, so a `true` output + // for every block is the proof that the intrinsic is wired correctly. + { + name: "context intrinsics expose the clock", + startBlock: 0, + stage: 0, + moduleName: "assert_context_intrinsics_0", + stateBundleSize: 10, + manifestPath: "./testdata/complex_substreams/complex-substreams-v0.1.0.spkg", + + expectRemainingFiles: []string{ + assertContextIntrinsics0 + "/outputs/0000000000-0000000010.output", + }, + + mapOutputFilesToDeepInspectForKeys: map[string]map[uint64]any{ + assertContextIntrinsics0 + "/outputs/0000000000-0000000010.output": { + 0: &pbsubstreamstest.Boolean{Result: true}, + 1: &pbsubstreamstest.Boolean{Result: true}, + 2: &pbsubstreamstest.Boolean{Result: true}, + 3: &pbsubstreamstest.Boolean{Result: true}, + 4: &pbsubstreamstest.Boolean{Result: true}, + 5: &pbsubstreamstest.Boolean{Result: true}, + 6: &pbsubstreamstest.Boolean{Result: true}, + 7: &pbsubstreamstest.Boolean{Result: true}, + 8: &pbsubstreamstest.Boolean{Result: true}, + 9: &pbsubstreamstest.Boolean{Result: true}, + }, + }, + }, + // This test checks that a module receiving data from both a filtered map and a Clock // does not trigger on every block, even when the index is being created in the same run { diff --git a/wasm/call.go b/wasm/call.go index 0e7a466de..552727e19 100644 --- a/wasm/call.go +++ b/wasm/call.go @@ -23,6 +23,7 @@ import ( "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" ) var ErrWasmDeterministicExec = errors.New("wasm execution failed deterministically") @@ -166,6 +167,17 @@ func (c *Call) ReachedLogsMaxByteCount() bool { return c.LogsByteCount >= maxTotalLogsByteCount } +// DoClock backs the `context::clock` intrinsic, returning the block clock encoded as a +// `sf.substreams.v1.Clock` protobuf message. +func (c *Call) DoClock() []byte { + out, err := proto.Marshal(c.Clock) + if err != nil { + c.PanicNonDeterministicError(fmt.Errorf("marshalling clock: %w", err)) + } + + return out +} + func (c *Call) DoSet(ord uint64, key string, value []byte) { now := time.Now() c.validateSimple("set", pbsubstreams.Module_KindStore_UPDATE_POLICY_SET, key) diff --git a/wasm/context_test.go b/wasm/context_test.go new file mode 100644 index 000000000..3405d2913 --- /dev/null +++ b/wasm/context_test.go @@ -0,0 +1,25 @@ +package wasm + +import ( + "testing" + "time" + + pbsubstreams "github.com/streamingfast/substreams/pb/sf/substreams/v1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/timestamppb" +) + +func TestCall_DoClock(t *testing.T) { + clock := &pbsubstreams.Clock{ + Id: "block-10", + Number: 10, + Timestamp: timestamppb.New(time.Unix(1000, 0).UTC()), + } + call := &Call{ModuleName: "mod", Clock: clock} + + out := &pbsubstreams.Clock{} + require.NoError(t, proto.Unmarshal(call.DoClock(), out)) + assert.True(t, proto.Equal(clock, out), "expected %s, got %s", clock, out) +} diff --git a/wasm/registry.go b/wasm/registry.go index 71d41fc8b..906941a0e 100644 --- a/wasm/registry.go +++ b/wasm/registry.go @@ -34,6 +34,9 @@ func (r *Registry) registerWASMExtension(namespace string, importName string, ex if namespace == "logger" { panic("cannot extend 'logger' wasm namespace") } + if namespace == "context" { + panic("cannot extend 'context' wasm namespace") + } if r.Extensions == nil { r.Extensions = map[string]map[string]WASMExtension{} diff --git a/wasm/wasmtime/context_externs.go b/wasm/wasmtime/context_externs.go new file mode 100644 index 000000000..d9e93801b --- /dev/null +++ b/wasm/wasmtime/context_externs.go @@ -0,0 +1,24 @@ +package wasmtime + +import ( + "fmt" + + wasmtime "github.com/bytecodealliance/wasmtime-go/v41" +) + +// registerContextImports exposes the execution context a module runs in. Unlike the +// `state` getters the value always exists, so the import returns nothing and simply +// writes the payload (a `{ptr, len}` pair) at `outputPtr`. +func (i *instance) registerContextImports(linker *wasmtime.Linker) error { + if err := linker.FuncWrap("context", "clock", + func(outputPtr int32) { + if err := writeOutputToHeap(i, outputPtr, i.CurrentCall.DoClock()); err != nil { + i.CurrentCall.PanicDeterministicError("writing clock to heap: %w", err) + } + }, + ); err != nil { + return fmt.Errorf("registering clock import: %w", err) + } + + return nil +} diff --git a/wasm/wasmtime/instance.go b/wasm/wasmtime/instance.go index 47e0a2c70..abad3b8f2 100644 --- a/wasm/wasmtime/instance.go +++ b/wasm/wasmtime/instance.go @@ -89,6 +89,10 @@ func (i *instance) newImports() error { if err != nil { return fmt.Errorf("registering state imports: %w", err) } + err = i.registerContextImports(linker) + if err != nil { + return fmt.Errorf("registering context imports: %w", err) + } if err = linker.FuncWrap("env", "register_panic", func(msgPtr, msgLength int32, filenamePtr, filenameLength int32, lineNumber, columnNumber int32, caller *wasmtime.Caller) { diff --git a/wasm/wazero/context_hostmod.go b/wasm/wazero/context_hostmod.go new file mode 100644 index 000000000..d5ec8168e --- /dev/null +++ b/wasm/wazero/context_hostmod.go @@ -0,0 +1,29 @@ +package wazero + +import ( + "context" + + "github.com/tetratelabs/wazero/api" + + "github.com/streamingfast/substreams/wasm" +) + +// ContextFuncs exposes the execution context a module runs in. Unlike the `state` getters +// the value always exists, so the import returns nothing and simply writes the payload +// (a `{ptr, len}` pair) at `output_ptr`. +var ContextFuncs = []funcs{ + { + "clock", + []parm{i32}, + []parm{}, + api.GoModuleFunc(func(ctx context.Context, mod api.Module, stack []uint64) { + outputPtr := uint32(stack[0]) + call := wasm.FromContext(ctx) + inst := instanceFromContext(ctx) + + if err := writeOutputToHeap(ctx, inst, outputPtr, call.DoClock()); err != nil { + call.PanicDeterministicError("writing clock to heap: %w", err) + } + }), + }, +} diff --git a/wasm/wazero/module.go b/wasm/wazero/module.go index 07bcc33c9..5279fb443 100644 --- a/wasm/wazero/module.go +++ b/wasm/wazero/module.go @@ -82,7 +82,11 @@ func newModule(ctx context.Context, wasmCode []byte, wasmCodeType string, regist if err != nil { return nil, err } - hostModules = append(hostModules, envModule, stateModule, loggerModule) + contextModule, err := AddHostFunctions(ctx, runtime, "context", ContextFuncs) + if err != nil { + return nil, err + } + hostModules = append(hostModules, envModule, stateModule, loggerModule, contextModule) // TODO: where to `Close()` the `runtime` here? // One runtime per request?