Add reader-based replay entry points for non-filesystem callers

claude · andrewh · commit b4a8daaa4d8f · 2026-06-17T17:02:17.000+01:00
Extract the bodies of ScanRecording and ReplayRecording into io.Reader variants (ScanRecordingFrom, ReplayRecordingFrom) and make the path-based functions thin wrappers over them. This lets callers that hold a recording in memory rather than on disk — notably pkg/synth compiled to js/wasm in the browser — drive replay without an os.Open dependency. The serialization side (WriteRecording/ReadRecording) was already reader/writer-based; only scan and replay-emit were path-locked. Behavior is unchanged for existing CLI callers, which still pass a path. Doc comments on the reader variants call out the two-pass dependency: the scan must precede the replay, so a single non-seekable reader can't serve both — in-memory callers hand a fresh reader over the same bytes to each. Adds reader-based test coverage and confirms pkg/synth still builds under GOOS=js GOARCH=wasm. Fixes #216 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011drMdKCDHZe6ozhAaa7un9
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Reader-based replay entry points `ScanRecordingFrom(io.Reader)` and
+  `ReplayRecordingFrom(ctx, io.Reader, …)` so non-filesystem callers (such as
+  `pkg/synth` compiled to `js/wasm`) can replay in-memory recordings. The
+  existing path-based `ScanRecording`/`ReplayRecording` now delegate to them;
+  no behavior change for CLI callers. (#216)
+
 - Replay mode: `motel import --record <file>` writes a newline-delimited
   recording sidecar of the source traces alongside the inferred topology, and a
   `mode: replay` config (with `recording: <file>`) re-emits the recorded trace
diff --git a/pkg/synth/replay.go b/pkg/synth/replay.go
@@ -95,19 +95,19 @@ type RecordingInfo struct {
 	Spans    int
 }
 
-// ScanRecording reads a recording once to collect the data needed to set up
-// replay: the set of services (for provider/resource creation) and the global
-// earliest start time (for relative time-shifting).
-func ScanRecording(path string) (RecordingInfo, error) {
-	f, err := os.Open(path)
-	if err != nil {
-		return RecordingInfo{}, fmt.Errorf("opening recording: %w", err)
-	}
-	defer f.Close() //nolint:errcheck // read-only file, close error is not actionable
-
+// ScanRecordingFrom reads a recording from r once to collect the data needed to
+// set up replay: the set of services (for provider/resource creation) and the
+// global earliest start time (for relative time-shifting).
+//
+// Replay is a two-pass operation: this scan must run before ReplayRecordingFrom
+// so the earliest start and service set are known up front. A single
+// non-seekable reader cannot serve both passes, so callers holding the
+// recording in memory should hand a fresh reader (e.g. bytes.NewReader over the
+// same bytes) to each function.
+func ScanRecordingFrom(r io.Reader) (RecordingInfo, error) {
 	seen := make(map[string]struct{})
 	info := RecordingInfo{}
-	err = ReadRecording(f, func(t RecordedTrace) error {
+	err := ReadRecording(r, func(t RecordedTrace) error {
 		info.Traces++
 		for _, s := range t.Spans {
 			info.Spans++
@@ -128,6 +128,17 @@ func ScanRecording(path string) (RecordingInfo, error) {
 	return info, nil
 }
 
+// ScanRecording reads a recording from the file at path once to collect the
+// data needed to set up replay. It is a thin wrapper over ScanRecordingFrom.
+func ScanRecording(path string) (RecordingInfo, error) {
+	f, err := os.Open(path)
+	if err != nil {
+		return RecordingInfo{}, fmt.Errorf("opening recording: %w", err)
+	}
+	defer f.Close() //nolint:errcheck // read-only file, close error is not actionable
+	return ScanRecordingFrom(f)
+}
+
 // ReplayOptions controls how a recording is re-emitted.
 type ReplayOptions struct {
 	// Verbatim emits spans with their original recorded timestamps. When false
@@ -230,22 +241,21 @@ func randomReplaySpanID() trace.SpanID {
 	return sid
 }
 
-// ReplayRecording streams the recording at path and emits each trace through
+// ReplayRecordingFrom streams the recording from r and emits each trace through
 // the given tracers and observers. Timestamps follow opts. Returns aggregate
 // emission statistics.
-func ReplayRecording(ctx context.Context, path string, tracers TracerSource, observers []SpanObserver, opts ReplayOptions) (*Stats, error) {
-	f, err := os.Open(path)
-	if err != nil {
-		return nil, fmt.Errorf("opening recording: %w", err)
-	}
-	defer f.Close() //nolint:errcheck // read-only file, close error is not actionable
-
+//
+// Relative time-shifting (the default, non-Verbatim mode) needs opts.Start,
+// which comes from a prior ScanRecordingFrom pass over the same recording. A
+// single non-seekable reader cannot serve both passes, so in-memory callers
+// should hand a fresh reader over the same bytes to each function.
+func ReplayRecordingFrom(ctx context.Context, r io.Reader, tracers TracerSource, observers []SpanObserver, opts ReplayOptions) (*Stats, error) {
 	shift := opts.shift()
 	var stats Stats
 	var rstats realtimeStats
 	start := time.Now()
 
-	err = ReadRecording(f, func(t RecordedTrace) error {
+	err := ReadRecording(r, func(t RecordedTrace) error {
 		if ctx.Err() != nil {
 			return ctx.Err()
 		}
@@ -274,6 +284,18 @@ func ReplayRecording(ctx context.Context, path string, tracers TracerSource, obs
 	return &stats, nil
 }
 
+// ReplayRecording streams the recording at path and emits each trace through
+// the given tracers and observers. It is a thin wrapper over
+// ReplayRecordingFrom.
+func ReplayRecording(ctx context.Context, path string, tracers TracerSource, observers []SpanObserver, opts ReplayOptions) (*Stats, error) {
+	f, err := os.Open(path)
+	if err != nil {
+		return nil, fmt.Errorf("opening recording: %w", err)
+	}
+	defer f.Close() //nolint:errcheck // read-only file, close error is not actionable
+	return ReplayRecordingFrom(ctx, f, tracers, observers, opts)
+}
+
 // buildReplayPlans converts a recorded trace into ordered SpanPlans. Spans are
 // indexed in tree order (parents before children) so emission can resolve the
 // parent context, and each child's window is clamped to sit within its parent
diff --git a/pkg/synth/replay_test.go b/pkg/synth/replay_test.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"os"
+	"slices"
 	"strings"
 	"testing"
 	"time"
@@ -303,3 +304,105 @@ func TestReplayRecordingRejectsInvalidPreservedIDs(t *testing.T) {
 		t.Fatalf("error = %v, want invalid trace ID", err)
 	}
 }
+
+// recordingBytes serializes traces into the newline-delimited recording format,
+// mirroring an in-memory (non-filesystem) recording such as the WASM playground
+// holds.
+func recordingBytes(t *testing.T, traces ...RecordedTrace) []byte {
+	t.Helper()
+	var buf bytes.Buffer
+	w := NewRecordingWriter(&buf)
+	for _, tr := range traces {
+		if err := w.Write(tr); err != nil {
+			t.Fatalf("write recording: %v", err)
+		}
+	}
+	return buf.Bytes()
+}
+
+func TestScanRecordingFromCollectsServicesAndStart(t *testing.T) {
+	rec := sampleRecording()
+	data := recordingBytes(t, rec)
+
+	info, err := ScanRecordingFrom(bytes.NewReader(data))
+	if err != nil {
+		t.Fatalf("scan: %v", err)
+	}
+	if info.Traces != 1 || info.Spans != 2 {
+		t.Fatalf("info = %+v, want 1 trace / 2 spans", info)
+	}
+	if got, want := info.Services, []string{"api", "db"}; !slices.Equal(got, want) {
+		t.Errorf("services = %v, want %v", got, want)
+	}
+	if !info.Start.Equal(rec.Spans[0].Start) {
+		t.Errorf("start = %v, want %v", info.Start, rec.Spans[0].Start)
+	}
+}
+
+func TestReplayRecordingFromEmitsViaTracers(t *testing.T) {
+	data := recordingBytes(t, sampleRecording())
+
+	rec := newReplayObserver()
+	stats, err := ReplayRecordingFrom(context.Background(), bytes.NewReader(data), noopTracers(), []SpanObserver{rec}, ReplayOptions{Verbatim: true})
+	if err != nil {
+		t.Fatalf("replay: %v", err)
+	}
+	if stats.Traces != 1 || stats.Spans != 2 {
+		t.Fatalf("stats = %+v, want 1 trace / 2 spans", stats)
+	}
+	if stats.Errors != 1 {
+		t.Errorf("errors = %d, want 1", stats.Errors)
+	}
+	if rec.started != 2 || rec.observed != 2 {
+		t.Errorf("observer fired started=%d observed=%d, want 2/2", rec.started, rec.observed)
+	}
+}
+
+// TestReaderReplayTwoPass exercises the in-memory flow the reader variants
+// exist for: a single recording held as bytes, scanned then replayed by handing
+// a fresh reader to each pass, with the scan's Start driving relative shifting.
+func TestReaderReplayTwoPass(t *testing.T) {
+	rec := sampleRecording()
+	data := recordingBytes(t, rec)
+	anchor := time.Date(2026, 6, 17, 0, 0, 0, 0, time.UTC)
+
+	info, err := ScanRecordingFrom(bytes.NewReader(data))
+	if err != nil {
+		t.Fatalf("scan: %v", err)
+	}
+
+	recorder := tracetest.NewSpanRecorder()
+	provider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(recorder))
+	t.Cleanup(func() {
+		if err := provider.Shutdown(context.Background()); err != nil {
+			t.Fatalf("shutdown tracer provider: %v", err)
+		}
+	})
+	tracers := func(name string) trace.Tracer { return provider.Tracer(name) }
+
+	stats, err := ReplayRecordingFrom(context.Background(), bytes.NewReader(data), tracers, nil, ReplayOptions{
+		Start:  info.Start,
+		Anchor: anchor,
+	})
+	if err != nil {
+		t.Fatalf("replay: %v", err)
+	}
+	if stats.Traces != 1 || stats.Spans != 2 {
+		t.Fatalf("stats = %+v, want 1 trace / 2 spans", stats)
+	}
+
+	spans := recorder.Ended()
+	if len(spans) != 2 {
+		t.Fatalf("got %d ended spans, want 2", len(spans))
+	}
+	// Relative mode maps the earliest recorded start onto the anchor.
+	var earliest time.Time
+	for _, s := range spans {
+		if earliest.IsZero() || s.StartTime().Before(earliest) {
+			earliest = s.StartTime()
+		}
+	}
+	if !earliest.Equal(anchor) {
+		t.Errorf("earliest replayed start = %v, want anchor %v", earliest, anchor)
+	}
+}
