diff --git a/docs/moq.md b/docs/moq.md
index d1b5e0a6..5dfdcc5e 100644
--- a/docs/moq.md
+++ b/docs/moq.md
@@ -98,9 +98,15 @@ Recording a stream to S3, driving its multistream targets and transcoding
 its renditions must each happen on exactly one node, however many carry the
 stream. Every node with a session for a stream, whether it ingested the
 segments or received them from a peer, contends for each job it is equipped
-to do (S3 configured, a Livepeer gateway, always for multistream) through a
-lease in the station's shared statedb (`leases`; `pkg/director/roles.go`).
-The holder renews the lease every 2 seconds; a node that dies stops, and
-the next contender takes the job over once the 6 second TTL has lapsed. A
-node that ends a session cleanly releases its leases, so a peer takes over
-at once. A node on its own statedb simply holds every lease.
+to do (S3 configured for recording, always for multistream) through a lease
+in the station's shared statedb (`leases`; `pkg/director/roles.go`). The
+holder renews the lease every 2 seconds; a node that dies stops, and the
+next contender takes the job over once the 6 second TTL has lapsed. A node
+that ends a session cleanly releases its leases, so a peer takes over at
+once. A node on its own statedb simply holds every lease.
+
+Transcoding is contended for only by a node that is an origin for the
+stream, one that ingests it or that a peer pushes it to, and that has a
+Livepeer gateway: peers take a stream's renditions from its origin, so
+renditions minted on a pull-only replica would reach nobody but its own
+viewers.
diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go
index db0196f8..7d86ee72 100644
--- a/pkg/cmd/streamplace.go
+++ b/pkg/cmd/streamplace.go
@@ -579,6 +579,11 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu
 	if err != nil {
 		return err
 	}
+	// Before any server builds its handler: getBroadcaster reports this
+	// URL, which is how a peer pushing streams here finds the listener.
+	if cli.MoqAddr != "" && wsrep != nil {
+		a.MoqURL = func() string { return websocketrep.AdvertisedMoqURL(cli, moqSelfSigned) }
+	}
 
 	ctx = log.WithLogValues(ctx, "version", build.Version)
 
@@ -650,7 +655,6 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu
 		default:
 			wsrep.MoqTLS = moqSelfSigned.TLSConfig()
 		}
-		a.MoqURL = func() string { return websocketrep.AdvertisedMoqURL(cli, moqSelfSigned) }
 	}
 
 	group.Go(func() error {
diff --git a/pkg/director/roles.go b/pkg/director/roles.go
index cbfaa298..30beea9e 100644
--- a/pkg/director/roles.go
+++ b/pkg/director/roles.go
@@ -40,9 +40,11 @@ type role struct {
 // Once per session, on the first segment of either provenance. For the
 // session's first segment the director runs Start in the background and
 // NewSegment right after it, so the two calls can race; sync.Once lets one
-// of them do the work and the other return.
-func (ss *StreamSession) startRoles(ctx context.Context, repoDID string) {
+// of them do the work and the other return. The loops run on the
+// session's context, so ending the session puts every job down.
+func (ss *StreamSession) startRoles(repoDID string) {
 	ss.roles.Do(func() {
+		ctx := ss.ctx
 		if ss.cli.S3Configured() {
 			ss.contend(ctx, repoDID, role{
 				name:  roleRecord,
@@ -57,67 +59,83 @@ func (ss *StreamSession) startRoles(ctx context.Context, repoDID string) {
 			},
 			stop: func() {},
 		})
-		if ss.cli.LivepeerGatewayURL != "" {
-			ss.contend(ctx, repoDID, role{
-				name:  roleTranscode,
-				start: func(context.Context) { ss.transcoding.Store(true) },
-				stop:  func() { ss.transcoding.Store(false) },
-			})
-		}
+	})
+}
+
+// startTranscodeRole enters the session into contention for transcoding,
+// once, and only on a node that is an origin for the stream: one that
+// ingests it, or that a peer pushes it to. Peers take a stream's
+// renditions from its origin, so renditions minted anywhere else would
+// reach nobody but that node's own viewers.
+func (ss *StreamSession) startTranscodeRole(repoDID string) {
+	if ss.cli.LivepeerGatewayURL == "" {
+		return
+	}
+	ss.transcodeRole.Do(func() {
+		ss.contend(ss.ctx, repoDID, role{
+			name:  roleTranscode,
+			start: func(context.Context) { ss.transcoding.Store(true) },
+			stop:  func() { ss.transcoding.Store(false) },
+		})
 	})
 }
 
 // contend runs one role's lease loop for the life of the session: take the
 // lease when it is free, renew it while held, give the job up when it is
-// lost, and release it on the way out so a peer takes over at once.
+// lost, and release it on the way out so a peer takes over at once. The
+// first check happens before contend returns, so a job this node wins is
+// up before the segment that started the contention is handled; the rest
+// run in the background.
 func (ss *StreamSession) contend(ctx context.Context, repoDID string, r role) {
 	name := "stream:" + r.name + ":" + repoDID
 	holder := ss.cli.ServerDID()
 	ctx = log.WithLogValues(ctx, "role", r.name)
-	ss.Go(ctx, func() error {
-		var held bool
-		var renewed time.Time
-		var cancel context.CancelFunc
-		drop := func() {
+	var held bool
+	var renewed time.Time
+	var cancel context.CancelFunc
+	check := func() {
+		ok, err := ss.statefulDB.AcquireLease(name, holder, roleLeaseTTL)
+		if err != nil {
+			// A database we cannot reach cannot have handed the lease to
+			// anyone either, unless this node alone is cut off: keep the
+			// job until the lease would have lapsed.
+			log.Warn(ctx, "stream role lease check failed", "error", err)
+			ok = held && time.Since(renewed) < roleLeaseTTL
+		} else if ok {
+			renewed = time.Now()
+		}
+		switch {
+		case ok && !held:
+			held = true
+			rctx, stop := context.WithCancel(ctx)
+			cancel = stop
+			log.Log(ctx, "taking up stream role")
+			r.start(rctx)
+		case !ok && held:
+			log.Log(ctx, "stream role taken over by another node")
 			held = false
 			cancel()
 			r.stop()
 		}
+	}
+	check()
+	ss.Go(ctx, func() error {
 		defer func() {
 			if held {
-				drop()
+				cancel()
+				r.stop()
 				if err := ss.statefulDB.ReleaseLease(name, holder); err != nil {
 					log.Warn(ctx, "could not release stream role lease", "error", err)
 				}
 			}
 		}()
 		for {
-			ok, err := ss.statefulDB.AcquireLease(name, holder, roleLeaseTTL)
-			if err != nil {
-				// A database we cannot reach cannot have handed the
-				// lease to anyone either, unless this node alone is cut
-				// off: keep the job until the lease would have lapsed.
-				log.Warn(ctx, "stream role lease check failed", "error", err)
-				ok = held && time.Since(renewed) < roleLeaseTTL
-			} else if ok {
-				renewed = time.Now()
-			}
-			switch {
-			case ok && !held:
-				held = true
-				rctx, stop := context.WithCancel(ctx)
-				cancel = stop
-				log.Log(ctx, "taking up stream role")
-				r.start(rctx)
-			case !ok && held:
-				log.Log(ctx, "stream role taken over by another node")
-				drop()
-			}
 			select {
 			case <-ctx.Done():
 				return nil
 			case <-time.After(roleLeaseRenew):
 			}
+			check()
 		}
 	})
 }
diff --git a/pkg/director/stream_session.go b/pkg/director/stream_session.go
index d2fb040e..91ce6388 100644
--- a/pkg/director/stream_session.go
+++ b/pkg/director/stream_session.go
@@ -78,7 +78,9 @@ type StreamSession struct {
 	s3Mu               sync.Mutex
 	s3Uploader         *s3.S3Uploader
 	// roles enters the session into the stream's jobs once; see startRoles.
-	roles sync.Once
+	// transcodeRole is separate: it starts on the first segment that makes
+	// this node an origin for the stream.
+	roles, transcodeRole sync.Once
 	// transcoding is whether this node holds the transcode role.
 	transcoding atomic.Bool
 }
@@ -143,7 +145,10 @@ func (ss *StreamSession) Start(ctx context.Context, notif *media.NewSegmentNotif
 	// Live recording, multistream targets and transcoding are jobs for
 	// exactly one node: every node carrying the stream contends for them,
 	// and the leases decide (see roles.go).
-	ss.startRoles(ctx, notif.Segment.RepoDID)
+	ss.startRoles(notif.Segment.RepoDID)
+	if notif.Local {
+		ss.startTranscodeRole(notif.Segment.RepoDID)
+	}
 
 	// Start background workers for status, origin, and livestream updates
 	ss.g.Go(func() error {
@@ -250,8 +255,12 @@ func (ss *StreamSession) NewSegment(ctx context.Context, notif *media.NewSegment
 	// A session that began with replicated segments (this node syndicated
 	// the stream) and then receives local ones is already in the running
 	// for the stream's jobs: contention starts on the first segment of
-	// either kind.
-	ss.startRoles(ctx, spseg.Creator)
+	// either kind. Transcoding is contended for only once this node is an
+	// origin for the stream (see startTranscodeRole).
+	ss.startRoles(spseg.Creator)
+	if notif.Local || ss.replicator.ReceivesPush(spseg.Creator) {
+		ss.startTranscodeRole(spseg.Creator)
+	}
 
 	// stream.received is enqueued once per StreamSession when the first local media segment is accepted.
 	if notif.Local && !ss.streamReceivedNotified {
@@ -329,7 +338,7 @@ func (ss *StreamSession) NewSegment(ctx context.Context, notif *media.NewSegment
 	}
 	ss.UpdateViewCount(ctx)
 	// Transcoding is a job for the node holding the transcode role; every
-	// other node carrying the stream receives the renditions with it.
+	// other node carrying the stream receives the renditions from it.
 	if ss.cli.LivepeerGatewayURL != "" && ss.transcoding.Load() {
 		// Claimed in arrival order: the renditions publish in it however
 		// the gateway's round trips finish.
diff --git a/pkg/moq/announce.go b/pkg/moq/announce.go
index 9164d1c4..3d8c51c2 100644
--- a/pkg/moq/announce.go
+++ b/pkg/moq/announce.go
@@ -17,15 +17,27 @@ type Announced struct {
 	st     bidiStream
 	prefix string
 
-	mu     sync.Mutex
-	err    error
-	paths  []string
+	mu  sync.Mutex
+	err error
+	// active is the route each live Announce ID names ("" for one that
+	// looped back through us and was discarded); ids are assigned in order
+	// and never reused, so count is the next (draft §5.1.1). Only live ids
+	// are kept, so a peer cycling a route costs nothing.
+	active map[uint64]string
+	byPath map[string]uint64
+	count  uint64
 	out    chan Announcement
 	done   chan struct{}
 	closed chan struct{}
 	once   sync.Once
 }
 
+// maxAnnounced bounds the routes a peer may have advertised at once. A
+// node ingests a handful of streams; one announcing thousands is not
+// describing broadcasts, and is cut off rather than allowed to grow our
+// state.
+const maxAnnounced = 1024
+
 // Announced asks the peer for the broadcasts under prefix. It does not
 // wait for the answer: a refusal (a reset) surfaces from Next.
 func (s *Session) Announced(ctx context.Context, prefix string) (*Announced, error) {
@@ -37,7 +49,7 @@ func (s *Session) Announced(ctx context.Context, prefix string) (*Announced, err
 		reset(st, errCancelled)
 		return nil, fmt.Errorf("writing announce request: %w", err)
 	}
-	a := &Announced{s: s, st: st, prefix: prefix, out: make(chan Announcement, 16), done: make(chan struct{}), closed: make(chan struct{})}
+	a := &Announced{s: s, st: st, prefix: prefix, active: map[uint64]string{}, byPath: map[string]uint64{}, out: make(chan Announcement, 16), done: make(chan struct{}), closed: make(chan struct{})}
 	go a.read(bufio.NewReader(st))
 	return a, nil
 }
@@ -136,19 +148,28 @@ func (a *Announced) read(br *bufio.Reader) {
 			}
 			path := a.prefix + suffix
 			// A route advertised twice is a protocol violation; one that
-			// looped back through us is discarded, id and all
+			// looped back through us is discarded, keeping only its id
 			// (draft §5.1.1.1).
-			if slices.Contains(a.paths, path) {
+			if _, dup := a.byPath[path]; dup {
 				err := fmt.Errorf("route %q announced twice", path)
 				a.s.fail(sessionProtocolViolation, err)
 				a.finish(err)
 				return
 			}
+			if len(a.active) >= maxAnnounced {
+				err := fmt.Errorf("peer has %d routes announced", len(a.active))
+				a.s.fail(sessionInternalError, err)
+				a.finish(err)
+				return
+			}
+			id := a.count
+			a.count++
 			if slices.Contains(hops, a.s.hop) || peerHop == a.s.hop {
-				a.paths = append(a.paths, "")
+				a.active[id] = ""
 				continue
 			}
-			a.paths = append(a.paths, path)
+			a.active[id] = path
+			a.byPath[path] = id
 			if !a.emit(Announcement{Path: path, Active: true}) {
 				return
 			}
@@ -159,21 +180,33 @@ func (a *Announced) read(br *bufio.Reader) {
 				a.finish(err)
 				return
 			}
-			if id >= uint64(len(a.paths)) {
-				err := fmt.Errorf("announce end for unknown id %d", id)
+			path, ok := a.active[id]
+			if !ok {
+				err := fmt.Errorf("announce end for unknown or retired id %d", id)
 				a.s.fail(sessionProtocolViolation, err)
 				a.finish(err)
 				return
 			}
-			path := a.paths[id]
-			a.paths[id] = ""
+			delete(a.active, id)
+			delete(a.byPath, path)
 			if path != "" && !a.emit(Announcement{Path: path}) {
 				return
 			}
 		case announceUpdate:
 			// New hops or cost for a route we already have: nothing to
-			// act on, but it must still decode.
-			b.varint()
+			// act on, but it must name a live id and decode.
+			id := b.varint()
+			if _, err := a.hops(&b); err != nil {
+				a.s.fail(sessionProtocolViolation, err)
+				a.finish(err)
+				return
+			}
+			if _, ok := a.active[id]; !ok {
+				err := fmt.Errorf("announce update for unknown or retired id %d", id)
+				a.s.fail(sessionProtocolViolation, err)
+				a.finish(err)
+				return
+			}
 			if _, err := a.hops(&b); err != nil {
 				a.s.fail(sessionProtocolViolation, err)
 				a.finish(err)
diff --git a/pkg/moq/wire.go b/pkg/moq/wire.go
index 7227d16c..61050626 100644
--- a/pkg/moq/wire.go
+++ b/pkg/moq/wire.go
@@ -67,6 +67,7 @@ const (
 // Session error codes (CONNECTION_CLOSE).
 const (
 	sessionNoError           = 0x0
+	sessionInternalError     = 0x1
 	sessionUnauthorized      = 0x2
 	sessionProtocolViolation = 0x3
 )
diff --git a/pkg/replication/iroh_replicator/kv.go b/pkg/replication/iroh_replicator/kv.go
index 5b2a7be0..e4bf58dc 100644
--- a/pkg/replication/iroh_replicator/kv.go
+++ b/pkg/replication/iroh_replicator/kv.go
@@ -126,6 +126,11 @@ func (swarm *IrohSwarm) BuildOriginRecord(origin *placestream.BroadcastOrigin) e
 	return nil
 }
 
+// ReceivesPush: the swarm has no push; a stream is local or gossiped.
+func (swarm *IrohSwarm) ReceivesPush(string) bool {
+	return false
+}
+
 func (swarm *IrohSwarm) Start(ctx context.Context, cli *config.CLI) error {
 	if len(cli.Tickets) > 0 {
 		err := swarm.Node.JoinPeers(cli.Tickets)
diff --git a/pkg/replication/replicator.go b/pkg/replication/replicator.go
index fef1ba31..c8573570 100644
--- a/pkg/replication/replicator.go
+++ b/pkg/replication/replicator.go
@@ -15,4 +15,8 @@ type Replicator interface {
 	SendSegment(context.Context, *media.NewSegmentNotification) error
 	// populate this origin record with whatever fields are pertinent to your replicator
 	BuildOriginRecord(*placestream.BroadcastOrigin) error
+	// ReceivesPush reports whether a peer is pushing streamer's stream to
+	// this node right now, which makes this node an origin for it: peers
+	// take the stream, renditions included, from here.
+	ReceivesPush(streamer string) bool
 }
diff --git a/pkg/replication/websocketrep/moq.go b/pkg/replication/websocketrep/moq.go
index f48a5020..a3883e0e 100644
--- a/pkg/replication/websocketrep/moq.go
+++ b/pkg/replication/websocketrep/moq.go
@@ -373,10 +373,12 @@ func (r *WebsocketReplicator) dialMoq(ctx context.Context, rawURL string, pub mo
 // publisher, and its announcements say which streams it is ingesting. Each
 // one is subscribed to and fed through the same validation as a pull;
 // anything that fails validation ends the session, the wire carrying MUXL
-// and nothing else. While a pushed stream flows this node also puts
-// itself in the station's broadcast_origins for it, so peers sharing the
-// statedb pull it from here: the pusher is typically behind NAT, and its
-// own origin record names a listener nobody can reach.
+// and nothing else. While a pushed stream's segments validate this node
+// also puts itself in the station's broadcast_origins for it, so peers
+// sharing the statedb pull it from here: the pusher is typically behind
+// NAT, and its own origin record names a listener nobody can reach. The
+// row follows validated segments, never the announcement alone: naming a
+// DID proves nothing, a signed segment does.
 func (r *WebsocketReplicator) acceptPush(ctx context.Context, sess *moq.Session) {
 	ann, err := sess.Announced(ctx, "")
 	if err != nil {
@@ -414,19 +416,24 @@ func (r *WebsocketReplicator) acceptPush(ctx context.Context, sess *moq.Session)
 		running[a.Path] = cancel
 		streamer := a.Path
 		g.Go(func() error {
-			err := r.consume(bctx, sess, streamer)
+			r.pushedMu.Lock()
+			r.pushed[streamer]++
+			r.pushedMu.Unlock()
+			defer func() {
+				r.pushedMu.Lock()
+				r.pushed[streamer]--
+				if r.pushed[streamer] == 0 {
+					delete(r.pushed, streamer)
+				}
+				r.pushedMu.Unlock()
+			}()
+			err := r.consume(bctx, sess, streamer, r.originAdvertiser(bctx, streamer))
 			if bctx.Err() != nil {
 				// Retracted by the peer, or the session is over.
 				return nil
 			}
 			return err
 		})
-		if r.state != nil {
-			g.Go(func() error {
-				r.advertiseOrigin(bctx, streamer)
-				return nil
-			})
-		}
 	}
 	for _, stop := range running {
 		stop()
@@ -436,27 +443,41 @@ func (r *WebsocketReplicator) acceptPush(ctx context.Context, sess *moq.Session)
 	}
 }
 
+// ReceivesPush reports whether a peer is pushing streamer's stream here.
+func (r *WebsocketReplicator) ReceivesPush(streamer string) bool {
+	r.pushedMu.Lock()
+	defer r.pushedMu.Unlock()
+	return r.pushed[streamer] > 0
+}
+
 // originRefresh is how often a pushed stream's broadcast_origins row is
-// touched; peers read the table every originPollInterval.
+// touched while its segments validate; peers read the table every
+// originPollInterval.
 var originRefresh = 5 * time.Second
 
-func (r *WebsocketReplicator) advertiseOrigin(ctx context.Context, streamer string) {
-	for {
-		if err := r.state.UpsertBroadcastOrigin(streamer, r.cli.ServerDID(), r.getMyMoqURL(), time.Now()); err != nil {
-			log.Warn(ctx, "moq push: could not record this node as origin", "streamer", streamer, "error", err)
-		}
-		select {
-		case <-ctx.Done():
+// originAdvertiser returns a per-validated-segment hook that keeps this
+// node's broadcast_origins row for streamer fresh; nil with no statedb.
+func (r *WebsocketReplicator) originAdvertiser(ctx context.Context, streamer string) func() {
+	if r.state == nil {
+		return nil
+	}
+	var last time.Time
+	return func() {
+		if time.Since(last) < originRefresh {
 			return
-		case <-time.After(originRefresh):
+		}
+		last = time.Now()
+		if err := r.state.UpsertBroadcastOrigin(streamer, r.cli.ServerDID(), r.getMyMoqURL(), last); err != nil {
+			log.Warn(ctx, "moq push: could not record this node as origin", "streamer", streamer, "error", err)
 		}
 	}
 }
 
 // consume subscribes to one streamer's source and rendition tracks on a
 // session and feeds them through validation until the session or ctx ends;
-// it always returns an error saying why.
-func (r *WebsocketReplicator) consume(ctx context.Context, sess *moq.Session, streamer string) error {
+// it always returns an error saying why. onSource, if set, runs after each
+// source segment validates.
+func (r *WebsocketReplicator) consume(ctx context.Context, sess *moq.Session, streamer string, onSource func()) error {
 	source, err := sess.Subscribe(ctx, streamer, MoqSourceTrack)
 	if err != nil {
 		spmetrics.ReplicationConnectErrorsTotal.Inc()
@@ -481,6 +502,12 @@ func (r *WebsocketReplicator) consume(ctx context.Context, sess *moq.Session, st
 			if err := r.mm.ValidateMP4(context.Background(), bytes.NewReader(f.Payload), false); err != nil {
 				return fmt.Errorf("could not validate segment: %w", err)
 			}
+			if onSource != nil {
+				onSource()
+			}
+			if err := r.mm.ValidateMP4(context.Background(), bytes.NewReader(f.Payload), false); err != nil {
+				return fmt.Errorf("could not validate segment: %w", err)
+			}
 		}
 	})
 	g.Go(func() error {
diff --git a/pkg/replication/websocketrep/websocket_replicator.go b/pkg/replication/websocketrep/websocket_replicator.go
index fe264ffd..c06bb5eb 100644
--- a/pkg/replication/websocketrep/websocket_replicator.go
+++ b/pkg/replication/websocketrep/websocket_replicator.go
@@ -52,6 +52,9 @@ type WebsocketReplicator struct {
 	// same with this node's live streams announced, for sessions to --peers.
 	publisher     *moqPublisher
 	pushPublisher *pushPublisher
+	// pushed counts the sessions currently pushing each streamer here.
+	pushedMu sync.Mutex
+	pushed   map[string]int
 }
 
 func NewWebsocketReplicator(cli *config.CLI, bus *bus.Bus, mod model.Model, mm *media.MediaManager, state *statedb.StatefulDB, moqCert *moq.SelfSignedCert) *WebsocketReplicator {
@@ -68,6 +71,7 @@ func NewWebsocketReplicator(cli *config.CLI, bus *bus.Bus, mod model.Model, mm *
 		moqCert:       moqCert,
 		publisher:     publisher,
 		pushPublisher: &pushPublisher{moqPublisher: publisher, live: newLiveSet(cli.StreamSessionTimeout)},
+		pushed:        map[string]int{},
 	}
 }
 
@@ -369,7 +373,7 @@ func (r *WebsocketReplicator) openMoq(ctx context.Context, origin *placestream.B
 	}
 	defer sess.Close()
 	log.Log(ctx, "syndication: connected to origin (moq)", "origin", *origin.MoqURL)
-	return r.consume(ctx, sess, origin.Streamer)
+	return r.consume(ctx, sess, origin.Streamer, nil)
 }
 
 // feedRenditions folds a rendition addendum the origin minted into this