refactor(sink): trim per-emit allocations and drop inert surface

Manifold.Attach is now copy-on-write so Sink reads the outlet slice without
copying it on every emit. The http Post builds its request body with
bytes.NewReader instead of copying through a string. CloudWatch gains a
clock-injectable PutLogEventAt so the timestamp is testable without the wall
clock. The unread WithPollerClock option and its stored-but-unused clock are
removed, and the PhaseTransform godoc now states it is reserved rather than
emitted by the built-in Emitter.

Signed-off-by: Joshua Temple <joshua.temple@stablekernel.com>

Author: joshua-temple

sink/cloudwatch/cloudwatch.go

@@ -42,10 +42,19 @@ func PutLogEvents(input *cloudwatchlogs.PutLogEventsInput) csink.Op[Client] {
 // group and stream. The event timestamp is set to the current time in
 // milliseconds since the Unix epoch, as required by the CloudWatch Logs API.
 // Callers that need to control the timestamp, sequence token, or multiple
-// events in one request should use [PutLogEvents] instead.
+// events in one request should use [PutLogEventAt] or [PutLogEvents] instead.
 func PutLogEvent(logGroup, logStream, message string) csink.Op[Client] {
+	return PutLogEventAt(logGroup, logStream, message, time.Now())
+}
+
+// PutLogEventAt returns an Op that sends a single log event stamped with the
+// supplied time. It is the deterministic, clock-injectable form of
+// [PutLogEvent]: tests pass a fixed time and assert the emitted timestamp
+// without depending on the wall clock. The time is converted to milliseconds
+// since the Unix epoch, as required by the CloudWatch Logs API.
+func PutLogEventAt(logGroup, logStream, message string, at time.Time) csink.Op[Client] {
 	return csink.OpFunc[Client](func(ctx context.Context, c Client) error {
-		ts := time.Now().UnixMilli()
+		ts := at.UnixMilli()
 		_, err := c.PutLogEvents(ctx, &cloudwatchlogs.PutLogEventsInput{
 			LogGroupName:  &logGroup,
 			LogStreamName: &logStream,

sink/cloudwatch/putlogevent_at_test.go

@@ -0,0 +1,59 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package cloudwatch_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	cw "github.com/stablekernel/crucible/sink/cloudwatch"
+)
+
+// TestPutLogEventAt_StampsSuppliedTime verifies the clock-injectable variant
+// stamps the event with the caller-supplied time rather than the wall clock, so
+// the emitted timestamp is deterministic.
+func TestPutLogEventAt_StampsSuppliedTime(t *testing.T) {
+	t.Parallel()
+
+	at := time.Unix(1700000000, 0) // a fixed, known instant
+	c := &fakeClient{}
+	op := cw.PutLogEventAt("/app/events", "app-stream", "deterministic", at)
+	if err := op.Apply(context.Background(), c); err != nil {
+		t.Fatalf("Apply() error = %v", err)
+	}
+	if len(c.calls) != 1 {
+		t.Fatalf("PutLogEvents call count = %d, want 1", len(c.calls))
+	}
+	events := c.calls[0].LogEvents
+	if len(events) != 1 {
+		t.Fatalf("events count = %d, want 1", len(events))
+	}
+	if events[0].Timestamp == nil {
+		t.Fatal("event timestamp is nil")
+	}
+	if got, want := *events[0].Timestamp, at.UnixMilli(); got != want {
+		t.Fatalf("timestamp = %d, want %d (the supplied time in ms)", got, want)
+	}
+	if events[0].Message == nil || *events[0].Message != "deterministic" {
+		t.Errorf("message = %v, want %q", events[0].Message, "deterministic")
+	}
+}
+
+// TestPutLogEvent_DelegatesToPutLogEventAt verifies the convenience wrapper
+// still emits exactly one event to the named group and stream.
+func TestPutLogEvent_DelegatesToPutLogEventAt(t *testing.T) {
+	t.Parallel()
+
+	c := &fakeClient{}
+	op := cw.PutLogEvent("/app/events", "app-stream", "now")
+	if err := op.Apply(context.Background(), c); err != nil {
+		t.Fatalf("Apply() error = %v", err)
+	}
+	if len(c.calls) != 1 || len(c.calls[0].LogEvents) != 1 {
+		t.Fatalf("calls = %+v, want one call with one event", c.calls)
+	}
+	if c.calls[0].LogEvents[0].Timestamp == nil {
+		t.Error("PutLogEvent should stamp a timestamp")
+	}
+}

sink/errors.go

@@ -18,7 +18,9 @@ type Phase string
 
 const (
 	// PhaseTransform marks a failure turning a payload into a destination
-	// operation (a registry transformer returning an error-bearing op).
+	// operation. Registry transformers cannot themselves return an error, so the
+	// built-in Emitter never emits this phase today; it is reserved for outlets
+	// whose payload-to-operation step can fail and want to classify it.
 	PhaseTransform Phase = "transform"
 	// PhaseApply marks a failure applying the operation to the destination
 	// client (the write itself).

sink/http/http.go

@@ -12,12 +12,12 @@
 package http
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
 	"io"
 	gohttp "net/http"
-	"strings"
 
 	csink "github.com/stablekernel/crucible/sink"
 )
@@ -35,7 +35,7 @@ type Doer interface {
 // A non-2xx status code is returned as an error that includes the status text.
 func Post(url, contentType string, body []byte) csink.Op[Doer] {
 	return csink.OpFunc[Doer](func(ctx context.Context, doer Doer) error {
-		req, err := gohttp.NewRequestWithContext(ctx, gohttp.MethodPost, url, strings.NewReader(string(body)))
+		req, err := gohttp.NewRequestWithContext(ctx, gohttp.MethodPost, url, bytes.NewReader(body))
 		if err != nil {
 			return fmt.Errorf("http sink: build request: %w", err)
 		}

sink/manifold.go

@@ -50,20 +50,26 @@ func NewManifold(opts ...Option) *Manifold {
 }
 
 // Attach adds outlets to the Manifold and returns it for chaining. It is safe
-// for concurrent use.
+// for concurrent use. Attach replaces the outlet slice with a freshly allocated
+// one (copy-on-write) rather than appending in place, so a concurrent Sink that
+// is iterating an earlier snapshot never observes a mutated backing array.
 func (m *Manifold) Attach(outlets ...Outlet) *Manifold {
 	m.mu.Lock()
 	defer m.mu.Unlock()
-	m.outlets = append(m.outlets, outlets...)
+	next := make([]Outlet, 0, len(m.outlets)+len(outlets))
+	next = append(next, m.outlets...)
+	next = append(next, outlets...)
+	m.outlets = next
 	return m
 }
 
-// snapshot returns a stable view of the attached outlets for one operation, so a
-// concurrent Attach never mutates the slice mid-fan-out.
+// snapshot returns the current outlet slice for one operation. Because Attach
+// is copy-on-write the returned slice is never mutated in place, so callers may
+// read it without copying; the RLock only guards reading the slice header.
 func (m *Manifold) snapshot() []Outlet {
 	m.mu.RLock()
 	defer m.mu.RUnlock()
-	return append([]Outlet(nil), m.outlets...)
+	return m.outlets
 }
 
 // Sink fans payload out to every attached outlet, fire-and-forget. It starts a

sink/manifold_concurrent_test.go

@@ -0,0 +1,35 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package sink_test
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"github.com/stablekernel/crucible/sink"
+)
+
+// TestManifoldConcurrentSinkAndAttach exercises the copy-on-write snapshot path:
+// Sink reads the outlet slice without copying it, so a concurrent Attach must
+// replace the slice rather than mutate the one a fan-out is iterating. Run under
+// -race, this fails if Attach ever mutates a slice a Sink is reading.
+func TestManifoldConcurrentSinkAndAttach(t *testing.T) {
+	t.Parallel()
+
+	m := sink.NewManifold().Attach(sink.NewBucket())
+
+	var wg sync.WaitGroup
+	for i := 0; i < 16; i++ {
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			m.Sink(context.Background(), "payload")
+		}()
+		go func() {
+			defer wg.Done()
+			m.Attach(sink.NewBucket())
+		}()
+	}
+	wg.Wait()
+}

sink/poller.go

@@ -18,51 +18,39 @@ type PollerOption func(*pollerConfig)
 
 type pollerConfig struct {
 	interval time.Duration
-	now      func() time.Time
 }
 
 func defaultPollerConfig() pollerConfig {
-	return pollerConfig{interval: 60 * time.Second, now: time.Now}
+	return pollerConfig{interval: 60 * time.Second}
 }
 
 // WithPollInterval sets the sampling period. The default is 60s.
 func WithPollInterval(d time.Duration) PollerOption {
 	return func(c *pollerConfig) { c.interval = d }
 }
 
-// WithPollerClock injects the clock the Poller reads, for deterministic tests.
-// The default is time.Now. A nil clock is ignored.
-func WithPollerClock(now func() time.Time) PollerOption {
-	return func(c *pollerConfig) {
-		if now != nil {
-			c.now = now
-		}
-	}
-}
-
 // Poller periodically runs a CollectFunc and sinks each yielded payload to a
 // target Outlet (via SinkBatch when the target supports it). Construct with
 // NewPoller; drive it with Start and stop it with Stop.
 type Poller struct {
 	target   Outlet
 	collect  CollectFunc
 	interval time.Duration
-	now      func() time.Time
 
 	mu      sync.Mutex
 	cancel  context.CancelFunc
 	done    chan struct{}
 	started bool
 }
 
-// NewPoller binds a target and a CollectFunc into a Poller. Defaults: interval
-// 60s, now = time.Now.
+// NewPoller binds a target and a CollectFunc into a Poller. The default
+// interval is 60s.
 func NewPoller(target Outlet, collect CollectFunc, opts ...PollerOption) *Poller {
 	cfg := defaultPollerConfig()
 	for _, o := range opts {
 		o(&cfg)
 	}
-	return &Poller{target: target, collect: collect, interval: cfg.interval, now: cfg.now}
+	return &Poller{target: target, collect: collect, interval: cfg.interval}
 }
 
 // Start launches the sampling loop and returns the Poller for chaining. It is