Review the time-domain program: centralize duplication, align refusals

Closing pass over the six landed slices, written by several hands:

- gate.go: the sync/playout gate twins shared ~100 copy-pasted lines —
  message-time extraction, stage naming, timebase refusals, and branch
  validation now live in one set of gate helpers; both files became
  thin wrappers over their policies.
- pipeline/readiness.go: the readiness countdown (arm/note/claim CAS
  logic and the ready-event text) was duplicated verbatim between the
  buffered and direct runners; one shared struct owns it now, each
  runner keeping only its loop-shape-specific walk.
- One bindGateDeps seam replaces the bindPlayoutClock/bindQoSReport
  pair at all three stage-materialization sites, and one
  markTimelinePaced helper replaces three hand-rolled assertions.
- Refusal drift fixed: rate now shares the timeline-control refusal
  gate with Pause/Resume — the unpaced refusal names the same three
  fixes, and rate on a closed task refuses with control.ErrNotRunning
  where it previously silently succeeded.
- Six copy-pasted sleep-trace assertion loops across the slice tests
  collapsed into one assertClockSleeps helper; no pin weakened.

Net -3 lines with ~200 lines of duplication centralized; zero exports
added; markdown delta zero. Hot-path audit confirmed every slice held
its budget (playout/sync admit windows, one-comparison QoS on-time
path, one atomic load for readiness, fully cold flush).

Gate: root build/vet/tests/staticcheck/gofmt, race subset, seek suites
at count=10, nested modules and all example module tests, doc pins.

Author: thesyncim

branch_compose_build.go

@@ -582,8 +582,7 @@ func (b *builder) newBranchComposeStepStageNamed(ctx context.Context, name strin
 		// Playout gates pace on the task timeline the runtime clone carries
 		// as its clock; binding here keeps the gate internal to the lowering.
 		// The QoS reporter rides the same clone so late admits reach Watch.
-		bindPlayoutClock(transform.stage, b.taskClock())
-		bindQoSReport(transform.stage, b.runtime.qosReportFunc())
+		bindGateDeps(transform.stage, b.taskClock(), b.runtime.qosReportFunc())
 		if name != "" && name != transform.stage.Name() {
 			return namedStage{name: name, stage: transform.stage}, stream, nil
 		}

gate.go

@@ -0,0 +1,93 @@
+package goav
+
+import (
+	"strings"
+	"time"
+
+	"github.com/thesyncim/goav/av"
+	"github.com/thesyncim/goav/errcode"
+	"github.com/thesyncim/goav/pipeline"
+)
+
+// This file holds what the two timed-gate families (sync, playout) share:
+// stage naming, the refusal for media without a usable PTS timebase, and the
+// pre-mutation validation that a stream carries timebase facts. The families
+// keep their own literal fix lists so refusals stay exact.
+
+// gateMessageTime extracts the stream and PTS a timed gate schedules on; ok
+// is false for media without a valid PTS timebase (and for events).
+func gateMessageTime(msg *pipeline.Message) (av.StreamID, time.Duration, bool) {
+	switch msg.Kind {
+	case pipeline.MessagePacket:
+		if msg.Packet == nil || !msg.Packet.PTS.Base.Valid() {
+			return "", 0, false
+		}
+		pts, ok := msg.Packet.PTS.ToDuration()
+		return msg.Packet.StreamID, pts, ok
+	case pipeline.MessageFrame:
+		if msg.Frame == nil || !msg.Frame.PTS.Base.Valid() {
+			return "", 0, false
+		}
+		pts, ok := msg.Frame.PTS.ToDuration()
+		return msg.Frame.StreamID, pts, ok
+	default:
+		return "", 0, false
+	}
+}
+
+// gateStageName builds the diagnostic node name for a timed gate:
+// prefix-<sanitized policy name>, with the prefix doubling as the default name.
+func gateStageName(prefix string, name string) string {
+	name = strings.TrimSpace(name)
+	if name == "" {
+		name = prefix
+	}
+	replacer := strings.NewReplacer(" ", "-", "/", "-", "\\", "-", "\t", "-", "\n", "-")
+	return prefix + "-" + replacer.Replace(name)
+}
+
+// gateTimebaseError refuses one media message that reached a timed gate
+// without a valid PTS timebase; the caller supplies its gate-specific fixes.
+func gateTimebaseError(operation string, node string, msg *pipeline.Message, fixes []string) error {
+	kind := ""
+	switch {
+	case msg == nil:
+	case msg.Packet != nil:
+		kind = "packet"
+	case msg.Frame != nil:
+		kind = "frame"
+	}
+	return &BuildError{
+		Phase:     phaseBuild,
+		Family:    errcode.FamilyForCode(runtimeBranchInvalidCode),
+		Code:      runtimeBranchInvalidCode,
+		Operation: operation,
+		Node:      node,
+		Reason:    "media message has no valid PTS timebase",
+		fields:    errDetails(errDetail("message", firstNonEmpty(kind, "unknown"))),
+		fixes:     buildErrorFixes(fixes),
+		cause:     errUnsupportedBuild,
+	}
+}
+
+// validateGatePolicyForBranch refuses, before graph mutation, a branch whose
+// stream declares no timebase facts a timed gate could schedule on; a valid
+// TimeBase or a Codec.ClockRate (RTP-style) satisfies it.
+func validateGatePolicyForBranch(operation string, branchName string, stream av.Stream, reason string, fixes []string) error {
+	if stream.TimeBase.Valid() {
+		return nil
+	}
+	if stream.Codec.ClockRate != 0 {
+		return nil
+	}
+	return &BuildError{
+		Phase:     phaseBuild,
+		Family:    errcode.FamilyForCode(runtimeBranchInvalidCode),
+		Code:      runtimeBranchInvalidCode,
+		Operation: operation,
+		Node:      firstNonEmpty(branchName, string(stream.ID), "branch"),
+		Reason:    reason,
+		fixes:     buildErrorFixes(fixes),
+		cause:     errUnsupportedBuild,
+	}
+}

join_plan.go

@@ -1462,8 +1462,7 @@ func (p *joinPlan) joinNodeBufferPolicy(policy pipeline.BufferPolicy, work workP
 
 // insertJoinArmStage appends a per-arm stage after upstream and returns its ref.
 func insertJoinArmStage(graph pipeline.Graph, rt *runtime, stage pipeline.Stage, upstream string) (string, error) {
-	bindPlayoutClock(stage, rt.clock)
-	bindQoSReport(stage, rt.qosReportFunc())
+	bindGateDeps(stage, rt.clock, rt.qosReportFunc())
 	ref, err := graph.AddStage(stage, rt.buffer)
 	if err != nil {
 		return "", err

pipeline/buffered.go

@@ -142,11 +142,8 @@ type bufferedRunner struct {
 	// topo is the immutable producer-side routing snapshot; emit reads it without
 	// g.mu. nil means closed.
 	topo atomic.Pointer[bufferedTopo]
-	// readyRemaining counts the sinks that had not yet received media when Run
-	// started; when it reaches zero the graph publishes av.EventTaskReady.
-	// readyFired guards the publish so readiness reports at most once.
-	readyRemaining atomic.Int64
-	readyFired     atomic.Bool
+	// ready is the once-per-run av.EventTaskReady countdown (readiness.go).
+	ready readiness
 }
 
 // rebuildTopoLocked publishes a fresh producer-side routing snapshot. Caller must
@@ -1120,9 +1117,7 @@ func (g *bufferedRunner) deliver(ctx context.Context, node *bufferedNode, msg *M
 }
 
 // armReadinessLocked snapshots the readiness set at Run start: the active
-// sinks that have not yet seen media. Caller must hold g.mu. With no sinks
-// to wait for there is nothing to report; with sinks all fed already the
-// graph is ready as it starts.
+// sinks that have not yet seen media. Caller must hold g.mu.
 func (g *bufferedRunner) armReadinessLocked() {
 	var sinks, unfed int64
 	for i := range g.nodes {
@@ -1135,38 +1130,32 @@ func (g *bufferedRunner) armReadinessLocked() {
 			unfed++
 		}
 	}
-	g.readyRemaining.Store(unfed)
-	if sinks > 0 && unfed == 0 {
+	if g.ready.arm(sinks, unfed) {
 		g.publishReady()
 	}
 }
 
-// noteSinkMedia counts a sink out of the readiness set exactly once (first
-// media delivery, or removal before any media). The CAS keeps the countdown
-// single-shot per sink; the last count publishes readiness.
+// noteSinkMedia counts a sink out of the readiness set (first media delivery,
+// or removal before any media); the last count publishes readiness.
 func (g *bufferedRunner) noteSinkMedia(sawMedia *atomic.Bool) {
-	if !sawMedia.CompareAndSwap(false, true) {
-		return
-	}
-	if g.readyRemaining.Add(-1) == 0 {
+	if g.ready.note(sawMedia) {
 		g.publishReady()
 	}
 }
 
 // publishReady reports av.EventTaskReady on the graph's observer stream, at
 // most once. Cold path: it runs once per graph, never per message.
 func (g *bufferedRunner) publishReady() {
-	if !g.readyFired.CompareAndSwap(false, true) {
+	if !g.ready.claim() {
 		return
 	}
-	event := av.Event{Type: av.EventTaskReady, Reason: "every sink received its first media message"}
 	g.eventsMu.Lock()
 	defer g.eventsMu.Unlock()
 	if g.eventsClosed {
 		return
 	}
 	select {
-	case g.events <- event:
+	case g.events <- taskReadyEvent():
 	default:
 		observeDrop(nil, DropObserver, &g.cold)
 	}

pipeline/direct.go

@@ -151,11 +151,8 @@ type directRunner struct {
 	eventsClosed bool
 	cold         coldStats
 	closed       bool
-	// readyRemaining counts the sinks that had not yet received media when Run
-	// started; when it reaches zero the graph publishes av.EventTaskReady.
-	// readyFired guards the publish so readiness reports at most once.
-	readyRemaining atomic.Int64
-	readyFired     atomic.Bool
+	// ready is the once-per-run av.EventTaskReady countdown (readiness.go).
+	ready readiness
 }
 
 // rebuildTopoLocked publishes a fresh immutable routing snapshot from the current
@@ -768,9 +765,7 @@ func (g *directRunner) deliverTopo(ctx context.Context, dst *directTopoNode, msg
 }
 
 // armReadinessRLocked snapshots the readiness set at Run start: the active
-// sinks that have not yet seen media. Caller must hold g.mu (read). With no
-// sinks to wait for there is nothing to report; with sinks all fed already
-// the graph is ready as it starts.
+// sinks that have not yet seen media. Caller must hold g.mu (read).
 func (g *directRunner) armReadinessRLocked() {
 	var sinks, unfed int64
 	for i := range g.nodes {
@@ -783,38 +778,32 @@ func (g *directRunner) armReadinessRLocked() {
 			unfed++
 		}
 	}
-	g.readyRemaining.Store(unfed)
-	if sinks > 0 && unfed == 0 {
+	if g.ready.arm(sinks, unfed) {
 		g.publishReady()
 	}
 }
 
-// noteSinkMedia counts a sink out of the readiness set exactly once (first
-// media delivery, or removal before any media). The CAS keeps the countdown
-// single-shot per sink; the last count publishes readiness.
+// noteSinkMedia counts a sink out of the readiness set (first media delivery,
+// or removal before any media); the last count publishes readiness.
 func (g *directRunner) noteSinkMedia(sawMedia *atomic.Bool) {
-	if !sawMedia.CompareAndSwap(false, true) {
-		return
-	}
-	if g.readyRemaining.Add(-1) == 0 {
+	if g.ready.note(sawMedia) {
 		g.publishReady()
 	}
 }
 
 // publishReady reports av.EventTaskReady on the graph's observer stream, at
 // most once. Cold path: it runs once per graph, never per message.
 func (g *directRunner) publishReady() {
-	if !g.readyFired.CompareAndSwap(false, true) {
+	if !g.ready.claim() {
 		return
 	}
-	event := av.Event{Type: av.EventTaskReady, Reason: "every sink received its first media message"}
 	g.eventsMu.Lock()
 	defer g.eventsMu.Unlock()
 	if g.eventsClosed {
 		return
 	}
 	select {
-	case g.events <- event:
+	case g.events <- taskReadyEvent():
 	default:
 		observeDrop(nil, DropObserver, &g.cold)
 	}

pipeline/readiness.go

@@ -0,0 +1,45 @@
+package pipeline
+
+import (
+	"sync/atomic"
+
+	"github.com/thesyncim/goav/av"
+)
+
+// readiness is the once-per-run countdown behind av.EventTaskReady, shared by
+// both runners: Run arms it with the sinks that have not yet seen media, the
+// sink delivery path counts each one out on its first packet or frame (Remove
+// counts out a sink that leaves unfed), and whoever takes the count to zero
+// publishes the event — at most once, guarded by fired.
+type readiness struct {
+	remaining atomic.Int64
+	fired     atomic.Bool
+}
+
+// arm snapshots the countdown at Run start and reports whether the graph is
+// ready as it starts: sinks exist and every one already saw media. With no
+// sinks there is nothing to wait for, so a sink-less graph never reports.
+func (r *readiness) arm(sinks, unfed int64) bool {
+	r.remaining.Store(unfed)
+	return sinks > 0 && unfed == 0
+}
+
+// note counts a sink out exactly once (the CAS keeps the countdown
+// single-shot per sink) and reports whether it was the last — the caller
+// publishes readiness.
+func (r *readiness) note(sawMedia *atomic.Bool) bool {
+	if !sawMedia.CompareAndSwap(false, true) {
+		return false
+	}
+	return r.remaining.Add(-1) == 0
+}
+
+// claim reports whether the caller wins the single readiness publish.
+func (r *readiness) claim() bool {
+	return r.fired.CompareAndSwap(false, true)
+}
+
+// taskReadyEvent is the graph-level av.EventTaskReady both runners publish.
+func taskReadyEvent() av.Event {
+	return av.Event{Type: av.EventTaskReady, Reason: "every sink received its first media message"}
+}