Add a context::clock WASM intrinsic

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.

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<u8> {
+        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<test::Boolean, Error> {
+    assert_eq!(context_intrinsics::get_clock(), clock);
+
+    Ok(test::Boolean { result: true })
+}

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");

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

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
 		{

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)

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)
+}

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{}

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
+}

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) {

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)
+			}
+		}),
+	},
+}

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?