Fix nil deref, swallowed cache read errors and log truncation cap

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 {

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

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

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

wasm/call.go

@@ -163,7 +163,7 @@ const maxTotalLogsByteCount = 5 * 1024 * 1024 // 5 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) {

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 5 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
+
+	// 20 messages = 5 MiB cumulative, reaching the total cap.
+	for i := 0; i < 20; 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, 21) // 20 messages + 1 truncation marker
+	require.Contains(t, c.Logs[20], "truncated")
+}