diff --git a/pipeline/cache/engine.go b/pipeline/cache/engine.go index a49fa2bf0..0ce4d9d61 100644 --- a/pipeline/cache/engine.go +++ b/pipeline/cache/engine.go @@ -53,12 +53,12 @@ func (e *Engine) NewBuffer(optionalBlock *pbbstream.Block, clock *pbsubstreams.C e.reversibleBuffers[clock.Number] = out for moduleName, existingExecOut := range e.existingExecOuts { val, ok, err := existingExecOut.Get(e.ctx, clock.Number) - if !ok { - continue - } if err != nil { return nil, fmt.Errorf("getting existing exec output for %s: %w", moduleName, err) } + if !ok { + continue + } err = out.Set(moduleName, val) if err != nil { diff --git a/pipeline/cache/engine_test.go b/pipeline/cache/engine_test.go new file mode 100644 index 000000000..50eebf394 --- /dev/null +++ b/pipeline/cache/engine_test.go @@ -0,0 +1,68 @@ +package cache + +import ( + "context" + "errors" + "iter" + "testing" + + pbsubstreams "github.com/streamingfast/substreams/pb/sf/substreams/v1" + "github.com/streamingfast/substreams/storage/execout" + pboutput "github.com/streamingfast/substreams/storage/execout/pb" + "github.com/stretchr/testify/require" +) + +// fakeFileReader implements execout.FileReader with canned Get results. +type fakeFileReader struct { + payload []byte + found bool + err error +} + +func (f *fakeFileReader) ReadNext() (*pboutput.Item, error) { return nil, nil } +func (f *fakeFileReader) Iter() iter.Seq2[*pboutput.Item, error] { + return func(yield func(*pboutput.Item, error) bool) {} +} +func (f *fakeFileReader) Get(ctx context.Context, blockNumber uint64) ([]byte, bool, error) { + return f.payload, f.found, f.err +} +func (f *fakeFileReader) ModuleName() string { return "test_module" } +func (f *fakeFileReader) Filename() string { return "test-file" } +func (f *fakeFileReader) Close() error { return nil } + +func TestEngineNewBuffer_ExistingExecOutReadErrorIsReturned(t *testing.T) { + readErr := errors.New("corrupted execout file") + engine, err := NewEngine(context.Background(), nil, "sf.test.v1.Block", map[string]execout.FileReader{ + "test_module": &fakeFileReader{err: readErr}, + }, nil) + require.NoError(t, err) + + _, err = engine.NewBuffer(nil, &pbsubstreams.Clock{Id: "block-1", Number: 1}, nil) + require.ErrorIs(t, err, readErr) +} + +func TestEngineNewBuffer_ExistingExecOutAbsentBlockIsSkipped(t *testing.T) { + engine, err := NewEngine(context.Background(), nil, "sf.test.v1.Block", map[string]execout.FileReader{ + "test_module": &fakeFileReader{found: false}, + }, nil) + require.NoError(t, err) + + buf, err := engine.NewBuffer(nil, &pbsubstreams.Clock{Id: "block-1", Number: 1}, nil) + require.NoError(t, err) + require.NotNil(t, buf) +} + +func TestEngineNewBuffer_ExistingExecOutFoundIsSet(t *testing.T) { + engine, err := NewEngine(context.Background(), nil, "sf.test.v1.Block", map[string]execout.FileReader{ + "test_module": &fakeFileReader{payload: []byte("payload"), found: true}, + }, nil) + require.NoError(t, err) + + buf, err := engine.NewBuffer(nil, &pbsubstreams.Clock{Id: "block-1", Number: 1}, nil) + require.NoError(t, err) + + val, cached, err := buf.Get("test_module") + require.NoError(t, err) + require.True(t, cached) + require.Equal(t, []byte("payload"), val) +} diff --git a/pipeline/exec/sharedcache.go b/pipeline/exec/sharedcache.go index 68fb27855..991adc78a 100644 --- a/pipeline/exec/sharedcache.go +++ b/pipeline/exec/sharedcache.go @@ -163,7 +163,9 @@ func (s *SharedCache) Execute( } inst, err := wasmModule.ExecuteNewCall(ctx, call, nil, wasmArguments, argValues) - inst.Close(ctx) + if inst != nil { + inst.Close(ctx) + } result.updateFromCall(call, err) result.metricsGatherer.ApplyToStats(reqctx.ReqStats(originalContext)) diff --git a/pipeline/exec/sharedcache_test.go b/pipeline/exec/sharedcache_test.go new file mode 100644 index 000000000..dc06ec173 --- /dev/null +++ b/pipeline/exec/sharedcache_test.go @@ -0,0 +1,51 @@ +package exec + +import ( + "context" + "errors" + "testing" + + "github.com/streamingfast/substreams/metrics" + pbsubstreams "github.com/streamingfast/substreams/pb/sf/substreams/v1" + "github.com/streamingfast/substreams/reqctx" + "github.com/streamingfast/substreams/wasm" + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +// failingModule simulates a wasm.Module whose ExecuteNewCall fails before an +// instance is created (e.g. wazero instantiation or writeToHeap failure), +// returning a nil instance alongside the error. +type failingModule struct { + err error +} + +func (f *failingModule) NewInstance(ctx context.Context) (wasm.Instance, error) { + return nil, errors.New("not implemented") +} + +func (f *failingModule) ExecuteNewCall(ctx context.Context, call *wasm.Call, cachedInstance wasm.Instance, arguments []wasm.Argument, argValues map[string][]byte) (wasm.Instance, error) { + return nil, f.err +} + +func (f *failingModule) Close(ctx context.Context) error { + return nil +} + +func TestSharedCacheExecute_ExecuteNewCallFailureReturnsErrorWithoutPanic(t *testing.T) { + stats := metrics.NewReqStats(&metrics.Config{}, nil, nil, zap.NewNop()) + ctx := reqctx.WithReqStats(context.Background(), stats) + ctx = reqctx.WithRequest(ctx, &reqctx.RequestDetails{UniqueID: 1}) + + clock := &pbsubstreams.Clock{Id: "block-1", Number: 1} + call := wasm.NewCall(ctx, clock, "test_module", "test_entrypoint", stats, nil, false, nil) + + sharedCache := NewSharedCache(10) + execErr := errors.New("could not instantiate wasm module: boom") + + var err error + require.NotPanics(t, func() { + err = sharedCache.Execute(ctx, &failingModule{err: execErr}, "modulehash", call, nil, nil, nil) + }) + require.ErrorIs(t, err, execErr) +} diff --git a/wasm/call.go b/wasm/call.go index db82965b8..0e7a466de 100644 --- a/wasm/call.go +++ b/wasm/call.go @@ -159,11 +159,11 @@ func (c *Call) SetOutputStore(store store.Store) { } const maxLogByteCount = 512 * 1024 // 512 KiB -const maxTotalLogsByteCount = 5 * 1024 * 1024 // 5 MiB +const maxTotalLogsByteCount = 2 * 1024 * 1024 // 2 MiB var maxTotalLogsByteCountHumanized = humanize.IBytes(maxTotalLogsByteCount) func (c *Call) ReachedLogsMaxByteCount() bool { - return c.LogsByteCount >= maxLogByteCount + return c.LogsByteCount >= maxTotalLogsByteCount } func (c *Call) DoSet(ord uint64, key string, value []byte) { diff --git a/wasm/call_logs_test.go b/wasm/call_logs_test.go new file mode 100644 index 000000000..f7e6e697d --- /dev/null +++ b/wasm/call_logs_test.go @@ -0,0 +1,42 @@ +package wasm + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestAppendLog_NotTruncatedBetweenPerMessageAndTotalCaps(t *testing.T) { + c := &Call{} + msg := strings.Repeat("a", 256*1024) // 256 KiB, under the 512 KiB per-message cap + + // 4 messages = 1 MiB cumulative: above the 512 KiB per-message cap, + // but well under the 2 MiB total cap. Nothing must be truncated. + for i := 0; i < 4; i++ { + c.AppendLog(msg) + } + + require.False(t, c.ReachedLogsMaxByteCount()) + require.False(t, c.logsTruncated) + require.Len(t, c.Logs, 4) + for _, log := range c.Logs { + require.NotContains(t, log, "truncated") + } +} + +func TestAppendLog_TruncatedAboveTotalCap(t *testing.T) { + c := &Call{} + msg := strings.Repeat("a", 256*1024) // 256 KiB per message + + // 8 messages = 2 MiB cumulative, reaching the total cap. + for i := 0; i < 8; i++ { + c.AppendLog(msg) + } + require.True(t, c.ReachedLogsMaxByteCount()) + + c.AppendLog("this must be dropped") + require.True(t, c.logsTruncated) + require.Len(t, c.Logs, 9) // 8 messages + 1 truncation marker + require.Contains(t, c.Logs[8], "truncated") +}