streamhub

package
v1.53.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 9, 2026 License: AGPL-3.0 Imports: 15 Imported by: 0

Documentation

Overview

Package streamhub provides the single-owner terminal output hub: one StreamHub per tmux session fans output out to N attached Subscribers over a Transport-agnostic interface, and is the sole caller of that session's resize/quiescence/capture-pane surface, eliminating the multi-connection resize/capture race that per-connection ownership allowed.

Index

Constants

View Source 
const (

	// DefaultHubTeardownGrace is how long a StreamHub with zero subscribers
	// waits before tearing down control mode, so a brief reconnect doesn't
	// kill it (Story 1.2.2).
	DefaultHubTeardownGrace = 5 * time.Second
)
View Source 
const MaxBatchWindow = 20 * time.Millisecond

MaxBatchWindow bounds how long a BatchWindow accumulates raw output bytes before flushing automatically with FlushCeiling — Story 2.1.1's AC. It is deliberately close to the latency ceiling server/services/connectrpc_websocket.go's existing per-connection coalesce loop already tolerates today (a 32-frame batch cap at typical control-mode output rates), so batching never regresses today's perceived latency.

Variables

View Source 
var ErrOwnershipResolvedToOtherPath = errors.New("streamhub: session ownership already resolved to a different StreamPath")

ErrOwnershipResolvedToOtherPath is returned by ResolveExpecting when the session's sticky StreamPath resolution does not match the caller's own intended path — i.e. the other side of the race won. Callers must treat this as an explicit signal to join the winning path (attach as a subscriber to its existing hub, or proceed as a legacy per-connection stream) rather than silently reinterpreting their own attempt as having succeeded (Task 3.1.2b).

View Source 
var ErrSessionNotStarted = errors.New("session not started or paused")

ErrSessionNotStarted is the sentinel a SessionController implementation returns from the methods below when the session simply hasn't finished starting yet (or is paused), as opposed to a dead/crashed controller. *session.Instance returns it during its cold-start window (session/instance_tmux.go). applyNegotiatedSize (hub.go) checks for it via errors.Is to skip-and-retry instead of tearing the whole hub down for a transient condition that resolves itself.

Functions

func ActiveHubs 

func ActiveHubs() int64

ActiveHubs returns the current count of live StreamHubs in this process — the same value streamhub_active_hubs' gauge callback observes.

func BatchFlushFramesCoalesced 

func BatchFlushFramesCoalesced() int64

BatchFlushFramesCoalesced returns the frames-coalesced count from the most recent batch flush across every hub in this process — the same value streamhub_batch_flush_frames_coalesced's histogram most recently recorded.

func OverlapInvariant 

func OverlapInvariant(sessionName string, ownerCount int)

OverlapInvariant is the production-reachable, load-bearing regression check named in plan.md's Domain Glossary: no two owners (legacy connection or hub) should ever hold resize/capture authority for one tmux session concurrently. StreamOwnershipLock.Resolve's mutex-guarded resolve-once-and-cache behavior already makes this structurally impossible under correct usage — this function is the defense-in-depth check that would surface a future regression in that guarantee immediately (e.g. HubRegistry.GetOrCreate calls it on every hub creation/lookup, Task 3.2's real production call site) rather than relying solely on code review to keep Resolve/GetOrCreate correct forever. It always emits slog.Error with full context and increments streamhub_overlap_invariant_violations_total — it never panics, since a panic in this single-operator daily-driver process would be worse than the bug it would catch. This is the one real implementation: earlier phases' overlapInvariantViolated/assertOverlapInvariant test-local helpers (Epic 1.4) have been retired in favor of calling this directly, via SetOverlapInvariantHookForTest for the t.Fatal-on-first-occurrence behavior every -race test in this plan requires.

func OverlapInvariantViolationsTotal 

func OverlapInvariantViolationsTotal() int64

OverlapInvariantViolationsTotal returns the process-wide count of OverlapInvariant violations detected so far — the same value streamhub_overlap_invariant_violations_total's counter observes. Must stay 0 across the whole dark-launch window (plan.md's Observability Plan / Success Metric).

func RegisterMetrics 

func RegisterMetrics() error

RegisterMetrics registers the six streamhub_* instruments named in plan.md's Observability Plan against the process's OTel MeterProvider (telemetry.GetMeter(), a delegating proxy safe to call before telemetry.Initialize — see session/unfinished.RegisterMetrics's identical pattern). Idempotent via sync.Once: package init already calls this once; it is exported so a caller (or a smoke test, matching session/unfinished/metrics_test.go) can call it again safely.

func ResizeNegotiationsTotal 

func ResizeNegotiationsTotal() int64

ResizeNegotiationsTotal returns the process-wide count of resize negotiations recorded so far (every RequestResize call from a CanResize-eligible subscriber, regardless of whether it actually changed the negotiated size) — the same total streamhub_resize_negotiations_total accumulates.

func SetOverlapInvariantHookForTest 

func SetOverlapInvariantHookForTest(hook func(sessionName string, ownerCount int))

SetOverlapInvariantHookForTest installs hook to run, in addition to OverlapInvariant's normal slog.Error+metric behavior, whenever a violation is detected. Pass nil to clear it. Tests should always clear it via t.Cleanup so a hook installed by one test can't leak into another running in the same package binary.

func SetSessionOverrideLookup 

func SetSessionOverrideLookup(lookup func(sessionName string) (forceHub bool, ok bool))

SetSessionOverrideLookup installs lookup as the per-session override source consulted by every StreamOwnershipLock's Resolve call. lookup should return (forceHub, true) when sessionName has an explicit override recorded, or (_, false) when it does not (falls back to the global flagValue). Pass nil to clear it — tests should always do so via t.Cleanup so a hook installed by one test can't leak into another running in the same package binary.

func SubscribersPerHub 

func SubscribersPerHub() int64

SubscribersPerHub returns the subscriber count observed on the most recent AttachSubscriber call across every hub in this process — the same value streamhub_subscribers_per_hub's histogram most recently recorded.

Types

type BatchWindow 

type BatchWindow struct {
	// contains filtered or unexported fields
}

BatchWindow is StreamHub's single, hub-owned opportunistic-with-ceiling accumulation buffer and timer (Story 2.1.1): every raw output event for one hub is appended here as an opaque byte range — concatenated verbatim, never truncated, reordered, or content-sniffed — and the accumulated bytes are handed to onFlush exactly once per flush regardless of how many subscribers are attached, either at the caller's next TryFlush (opportunistic) or after maxWindow elapses since the first buffered byte (ceiling), whichever comes first. Exactly one BatchWindow exists per hub, so exactly one accumulation buffer and one *time.Timer exist per hub too, never one per subscriber.

func NewBatchWindow 

func NewBatchWindow(onFlush func(BroadcastUnit), opts ...BatchWindowOption) *BatchWindow

NewBatchWindow constructs a BatchWindow that hands each flushed BroadcastUnit to onFlush. onFlush must not block indefinitely — it runs synchronously on whichever goroutine triggered the flush (an Add/TryFlush caller, or the ceiling timer's own goroutine).

func (*BatchWindow) Add 

func (b *BatchWindow) Add(data []byte)

Add appends data to the pending accumulation buffer as an opaque byte range. Arms the ceiling timer exactly once per window, the moment the buffer transitions from empty to non-empty; every subsequent Add call within the same still-pending window never re-arms it — the invariant that keeps "exactly one timer per hub" true across an arbitrarily long burst of events.

func (*BatchWindow) Bypass 

func (b *BatchWindow) Bypass(data []byte)

Bypass immediately delivers data as its own BroadcastUnit, stamped with the next HubSequenceNumber, without touching the pending accumulation buffer at all (Story 2.1.2) — ResizeQuiescence signals and the post-resize CatchUpSnapshot must never wait behind a pending batch's opportunistic-or-ceiling flush (research/pitfalls.md §2c/§2d). Any already-pending batch still flushes later on its own normal schedule; Bypass never cancels or folds into it.

func (*BatchWindow) FlushCount 

func (b *BatchWindow) FlushCount() int64

FlushCount reports how many times the accumulation/coalesce step (flushLocked) has executed — test-only instrumentation for Task 2.1.1e's call-counting AC (must be 1 for an N-subscriber burst, not N).

func (*BatchWindow) TimersArmed 

func (b *BatchWindow) TimersArmed() int64

TimersArmed reports how many times the ceiling timer has been armed — test-only instrumentation proving at most one timer exists per window regardless of how many Add calls or subscribers are involved.

func (*BatchWindow) TryFlush 

func (b *BatchWindow) TryFlush()

TryFlush flushes any pending accumulated bytes immediately, with reason FlushOpportunistic. This is the hook for "the next subscriber-write opportunity" (Story 2.1.1's AC): a caller feeding raw output into Add calls TryFlush once it has drained every immediately-available event, mirroring the `default: break coalesce` point in server/services/connectrpc_websocket.go's existing per-connection coalesce loop. A no-op if nothing is buffered.

type BatchWindowOption 

type BatchWindowOption func(*BatchWindow)

BatchWindowOption configures a BatchWindow at construction time.

func WithMaxBatchWindow 

func WithMaxBatchWindow(d time.Duration) BatchWindowOption

WithMaxBatchWindow overrides MaxBatchWindow — tests use this to shrink the ceiling so ceiling-flush assertions run in milliseconds.

type BroadcastUnit 

type BroadcastUnit struct {
	Seq    HubSequenceNumber
	Data   []byte
	Reason FlushReason

	// FramesCoalesced is the number of underlying Add calls folded into this
	// unit (Task 3.2.2a/Story 3.2.2's streamhub_batch_flush_frames_coalesced
	// metric) — 1 for a Bypass unit, since it never touches accumulation at
	// all and represents exactly one control/quiescence message.
	FramesCoalesced int
}

BroadcastUnit is one flushed batch or bypassed control message, stamped with the hub's HubSequenceNumber at the moment it is actually handed to onFlush (broadcast time) — never at Add/Bypass call time, so a unit that spent longer accumulating never receives a lower sequence number than one that bypassed it later (Story 2.1.2's AC).

type FlushReason 

type FlushReason int

FlushReason records why a BatchWindow flushed a batch, distinguishing an opportunistic flush (the caller found a subscriber-write opportunity before the ceiling elapsed) from one forced by MaxBatchWindow, and from a Bypass call that never went through accumulation at all. Every switch over FlushReason must include a default: panic("unhandled FlushReason") case, matching HubLifecycleState/StreamPath in types.go (enforced by the `exhaustive` linter). 

const (
	// FlushOpportunistic fires when the caller signals a subscriber-write
	// opportunity (TryFlush) while data is pending.
	FlushOpportunistic FlushReason = iota
	// FlushCeiling fires when MaxBatchWindow elapses since the first byte
	// buffered in the current window, with no opportunistic flush beating it.
	FlushCeiling
	// FlushBypass marks a unit that never touched the accumulation buffer at
	// all (Story 2.1.2) — a control/quiescence message sent via Bypass.
	FlushBypass
)

func (FlushReason) String 

func (r FlushReason) String() string

String renders FlushReason for logging.

type HubLifecycleState 

type HubLifecycleState int

HubLifecycleState is the exhaustive set of states a StreamHub can be in. Every switch over it must include a default: panic("unhandled HubLifecycleState") case so a new state can't silently fall through unhandled logic (enforced by the `exhaustive` linter). 

const (
	// HubStarting is the state between hub creation and its first
	// successful subscriber attach / control-mode start.
	HubStarting HubLifecycleState = iota

	// HubActive is the normal operating state: at least one subscriber is
	// attached and the hub is forwarding output.
	HubActive

	// HubDraining is entered when the last subscriber detaches; the hub
	// waits out a grace period before tearing down, so a brief reconnect
	// doesn't kill control mode.
	HubDraining

	// HubTornDown is the terminal state: control mode has been stopped and
	// the hub is no longer usable.
	HubTornDown
)

type HubOption 

type HubOption func(*StreamHub)

HubOption configures a StreamHub at construction time. Tests use these to shrink buffer sizes and grace periods so lifecycle/eviction assertions run in milliseconds instead of seconds; production callers can rely on the zero-value defaults below.

func WithBatchMaxWindow 

func WithBatchMaxWindow(d time.Duration) HubOption

WithBatchMaxWindow overrides the hub's BatchWindow ceiling (default: MaxBatchWindow, 20ms) — tests use this to shrink it so ceiling-flush assertions run in milliseconds.

func WithQuiescenceQuietPeriod 

func WithQuiescenceQuietPeriod(d time.Duration) HubOption

WithQuiescenceQuietPeriod overrides how long a resize's quiescence-wait must see no update before declaring the reflow settled (default: 200ms).

func WithQuiescenceTimeout 

func WithQuiescenceTimeout(d time.Duration) HubOption

WithQuiescenceTimeout overrides the hard deadline a resize's quiescence-wait gives up at (default: 500ms, matching the pre-hub per-connection behavior). Tests use this to shrink the deadline so timeout-path assertions run in milliseconds.

func WithSlowSubscriberGrace 

func WithSlowSubscriberGrace(d time.Duration) HubOption

WithSlowSubscriberGrace overrides how long a full outbound queue is tolerated before the subscriber is evicted.

func WithSubscriberBufferSize 

func WithSubscriberBufferSize(n int) HubOption

WithSubscriberBufferSize overrides the outbound queue depth for every subscriber attached after this option is applied.

func WithTeardownGrace 

func WithTeardownGrace(d time.Duration) HubOption

WithTeardownGrace overrides how long the hub waits after its last subscriber detaches before tearing down.

type HubSequenceNumber 

type HubSequenceNumber uint64

HubSequenceNumber is a monotonic, per-hub ordinal stamped on every broadcast unit (a flushed BatchWindow frame or a Bypass control message) at the moment it is actually broadcast, not when it was first buffered — giving every Subscriber the same total order regardless of its own flush cadence (plan.md's Domain Glossary; Story 2.1.2's AC).

type MemoryTransport 

type MemoryTransport struct {
	// contains filtered or unexported fields
}

MemoryTransport is an in-process Transport implementation (Story 1.4.1): it lets hub tests exercise attach/broadcast/eviction behavior without a real tmux process or network socket, the required Testability NFR. It records every frame it receives and can be configured, via With* options, to block or error on Send so tests can deterministically drive the slow-subscriber and Transport.Send-error eviction paths (Story 1.2.1, Story 1.4.2) instead of relying on real network timing.

func NewMemoryTransport 

func NewMemoryTransport(opts ...MemoryTransportOption) *MemoryTransport

NewMemoryTransport returns a MemoryTransport configured by opts. With no options, Send always succeeds immediately and records the frame.

func (*MemoryTransport) Close 

func (m *MemoryTransport) Close() error

Close implements Transport. It releases any Send call blocked by WithBlockingSend (treated the same as an explicit Unblock) and is safe to call more than once.

func (*MemoryTransport) IsClosed 

func (m *MemoryTransport) IsClosed() bool

IsClosed reports whether Close has been called.

func (*MemoryTransport) ReceivedFrames 

func (m *MemoryTransport) ReceivedFrames() [][]byte

ReceivedFrames returns a copy of every frame successfully delivered via Send so far, in delivery order.

func (*MemoryTransport) Send 

func (m *MemoryTransport) Send(data []byte) error

Send implements Transport. Behavior depends on how the transport was constructed: WithErrorSend returns its configured error immediately; WithBlockingSend blocks until Unblock or Close is called; otherwise the frame is recorded and Send returns nil.

func (*MemoryTransport) Unblock 

func (m *MemoryTransport) Unblock()

Unblock releases any Send call currently blocked by WithBlockingSend. It is safe to call even if no Send is currently blocked, and safe to call more than once.

type MemoryTransportOption 

type MemoryTransportOption func(*MemoryTransport)

MemoryTransportOption configures a MemoryTransport at construction time.

func WithBlockingSend 

func WithBlockingSend() MemoryTransportOption

WithBlockingSend makes every Send call block until Unblock is called (or the transport is closed), simulating a stalled writer — e.g. a dead network connection whose write never returns — so tests can deterministically exercise the slow-subscriber eviction path.

func WithErrorSend 

func WithErrorSend(err error) MemoryTransportOption

WithErrorSend makes every Send call return err instead of recording the frame, so tests can exercise the Transport.Send-error eviction path.

type RawPaneContent added in v1.49.0 

type RawPaneContent string

RawPaneContent is tmux `capture-pane -p -e` output captured WITHOUT -J: one output line per visual row of the pane, exactly matching what a terminal emulator displayed, with cursor-positioning/SGR escape codes intact. This is the only form safe to feed into prepareSnapshotContent and replay into a live terminal emulator (xterm.js) — see SessionController.CapturePaneContentRaw's doc comment.

It is deliberately NOT the type of tmux's -J ("joined") capture variant, which merges soft-wrapped continuation rows back into their original logical line for plain-text uses (search, logging, debug dumps) — safe to display as text, but replaying -J output into a terminal emulator destroys the visual grid structure and drops cursor codes the -J join strips as a side effect. Keeping that variant as an untyped string (rather than a matching JoinedPaneContent) is deliberate: prepareSnapshotContent requiring RawPaneContent is what makes passing the wrong variant a compile error, which is the actual safety property this type exists for — every -J consumer in this codebase treats its result as display/search text, never as renderer input, so there is no equivalent mistake to guard against on that side.

type ResizeVote 

type ResizeVote struct {
	SubscriberID SubscriberID
	Size         TerminalSize
}

ResizeVote is a {SubscriberID, TerminalSize} tuple submitted by a capability-eligible Subscriber toward the hub's NegotiatedSize (Task 1.3.1b).

type SessionController 

type SessionController interface {
	// SetWindowSizeContext propagates a negotiated resize to the underlying
	// tmux session, bounded by ctx. StreamHub.applyNegotiatedSize is its sole
	// caller.
	SetWindowSizeContext(ctx context.Context, cols, rows int) error

	// ResizePTY resizes the terminal dimensions, mirroring
	// *session.Instance.ResizePTY's cols/rows nudge pattern. Part of the
	// interface per Task 1.3.2a; Epic 1.3's resize pipeline does not call it
	// (that call site is the attach-time handshake nudge in
	// server/services/connectrpc_websocket.go, left in place until a later
	// phase migrates it onto the hub).
	ResizePTY(cols, rows int) error

	// CapturePaneContentRawContext captures the current visible pane content
	// without joining tmux's soft-wrapped lines (no -J) and with cursor
	// positioning codes intact, bounded by ctx. The hub calls this exactly
	// once per resize, after quiescence is reached, and once at attach time
	// for a new subscriber's catch-up snapshot; both call sites run the
	// result through prepareSnapshotContent/withCursorSync
	// (snapshot_prepare.go) before broadcasting. Deliberately not the joined
	// CapturePaneContent variant: -J strips the escape codes a snapshot's
	// cursor-sync depends on and collapses tmux's own wrap points, which is
	// what made every post-resize snapshot render staircased across the
	// previous frame (2026-08-25 reflow bug — see snapshot_prepare.go's doc
	// comment).
	CapturePaneContentRawContext(ctx context.Context) (RawPaneContent, error)

	// GetPaneCursorPosition reports the tmux pane's current cursor
	// coordinates, used by withCursorSync (snapshot_prepare.go) to reposition
	// the client cursor after a snapshot renders.
	GetPaneCursorPosition() (x, y int, err error)

	// StartControlMode ensures the underlying control-mode process is
	// running, forking a fresh one if it isn't (refcounted -- see
	// session/tmux/control_mode.go) and otherwise a no-op. Called by
	// pumpControlModeOutputIntoHub (server/services/connectrpc_websocket.go)
	// before every (re)subscribe, matching StopControlMode's one call from
	// StreamHub.ForceTeardown: without this, a control-mode crash mid-session
	// left the pump looping forever against a permanently pre-closed
	// subscription -- SubscribeControlModeUpdates alone never restarts a
	// dead process, only StartControlMode does -- silently and permanently
	// starving the hub (and every subscriber watching it) of live output,
	// including the subscriber's own typed input echoing back (2026-09-01).
	StartControlMode() error

	// StopControlMode stops the control-mode stream. Called exactly once by
	// StreamHub.ForceTeardown.
	StopControlMode() error

	// SubscribeControlModeUpdates registers a consumer of raw control-mode
	// output for this session, returning a subscriber ID and a read-only
	// channel of raw frames. StreamHub.applyNegotiatedSize uses this to
	// detect quiescence after a resize.
	SubscribeControlModeUpdates() (string, <-chan []byte)

	// UnsubscribeControlModeUpdates removes a subscription by the ID
	// returned from SubscribeControlModeUpdates. Implementations must close
	// the corresponding channel so a range loop reading from it can exit
	// without leaking. That close may be deferred past this call's return
	// (up to controlModeSlowSubscriberGrace) if a slow-drain is in flight
	// for this subscriber when Unsubscribe is called.
	UnsubscribeControlModeUpdates(id string)
}

SessionController is the narrow interface StreamHub depends on for resize/quiescence/capture/teardown, instead of the concrete *session.Instance type (Task 1.3.2a). session/streamhub never imports package session; *session.Instance satisfies SessionController structurally, which is what keeps package session's own dependency on session/streamhub (for StreamOwnershipLock, Story 3.1.2) a safe one-way edge rather than an import cycle — see plan.md's SessionController Pattern Decisions entry.

Scoped to exactly the seven *session.Instance methods it mirrors: session/instance_tmux.go's ResizePTY (:587), CapturePaneContentRawContext, GetPaneCursorPosition (:792), StopControlMode (:727), SubscribeControlModeUpdates (:733), UnsubscribeControlModeUpdates (:738), and SetWindowSizeContext.

SetWindowSizeContext/CapturePaneContentRawContext take ctx rather than reusing *session.Instance's existing context-less SetWindowSize/ CapturePaneContentRaw (which stay in place for their many other, non-hub callers) — a caller-supplied deadline lets applyNegotiatedSize's callers signal "stop waiting" early (e.g. the WebSocket connection whose resize vote triggered this call disconnecting) instead of only ever waiting out a fixed internal ceiling with no way to cancel sooner.

type StreamHub 

type StreamHub struct {
	// contains filtered or unexported fields
}

StreamHub is the single-owner runtime object for one tmux session's output stream: it fans output out to every attached subscriber over a Transport-agnostic interface, and is the sole caller of that session's resize/quiescence/capture-pane surface (Epic 1.3) via SessionController. Epic 1.2 implements the subscriber registry, fan-out, and lifecycle/teardown that this builds on.

func NewStreamHub 

func NewStreamHub(sessionName string, controller SessionController, opts ...HubOption) *StreamHub

NewStreamHub constructs a StreamHub for one tmux session, starting in HubStarting state with zero subscribers. controller is used by ForceTeardown and the resize/quiescence/capture pipeline; it may be nil in tests that never exercise either.

func (*StreamHub) AttachSubscriber 

func (h *StreamHub) AttachSubscriber(transport Transport, capability SubscriberCapability) SubscriberID

AttachSubscriber registers a new subscriber backed by transport, starts its writer goroutine, and returns its SubscriberID. Attaching during HubDraining cancels the pending teardown (Task 1.2.2a) and returns the hub to HubActive; attaching to a HubTornDown hub also reactivates it, since streamhub itself has no opinion on whether that's a legitimate reattach — that policy belongs to HubRegistry (Epic 3).

func (*StreamHub) Broadcast 

func (h *StreamHub) Broadcast(data []byte)

Broadcast fans data out to every attached subscriber via a non-blocking send. A subscriber whose outbound queue is observed full is never blocked on: the first observation arms a one-shot grace-period timer, independent of whether further frames are broadcast, and eviction (Task 1.2.1e) fires only if the queue is still full when that timer expires — a subscriber that merely had a momentary burst and drained in time is never evicted or re-armed for the same stall.

func (*StreamHub) DetachSubscriber 

func (h *StreamHub) DetachSubscriber(id SubscriberID)

DetachSubscriber removes the subscriber and stops its writer goroutine without leaking it. If this was the last subscriber, the hub schedules teardown after its grace period rather than tearing down immediately (Story 1.2.2). Detaching an unknown SubscriberID is a no-op.

func (*StreamHub) ForceTeardown 

func (h *StreamHub) ForceTeardown() error

ForceTeardown tears the hub down unconditionally: every remaining subscriber is closed, SessionController.StopControlMode() is invoked exactly once, and only then does the hub transition to HubTornDown — callers polling State() must never observe HubTornDown before StopControlMode has actually returned (see TestStreamHub_should_NotReportHubTornDown_While_StopControlModeInFlight). It is the single teardown code path reached both by grace-period expiry (onTeardownGraceExpired) and an external trigger, e.g. Story 3.1.2's flag-flip-to-legacy case. Safe to call concurrently or more than once: teardownInFlight, not h.state, rejects a second caller, since h.state can't serve as that guard until the transition completes.

func (*StreamHub) MarkPumpExited added in v1.49.0 

func (h *StreamHub) MarkPumpExited()

MarkPumpExited records that this hub's raw-output pump has stopped running — called by pumpControlModeOutputIntoHub immediately before each of its return points, so TryStartPump's next caller can correctly restart one. See TryStartPump's doc comment.

func (*StreamHub) NegotiatedSize 

func (h *StreamHub) NegotiatedSize() TerminalSize

NegotiatedSize returns the hub's current resolved TerminalSize: the component-wise minimum across every CanResize subscriber that has ever called RequestResize (ADR-002's smallest-common-size model). It is the zero TerminalSize until the first accepted vote.

func (*StreamHub) OnRawOutput 

func (h *StreamHub) OnRawOutput(data []byte)

OnRawOutput is StreamHub's gate for raw tmux output once a caller (the production wiring that connects a StreamHub to its tmux session, outside Epic 1.3's scope) drives it with SessionController.SubscribeControlModeUpdates frames. It always counts the arrival toward quiescence via the resizing flag's readers, but suppresses broadcasting the frame itself while a resize is in progress (Task 1.3.2d) — mirroring server/services/connectrpc_websocket.go's existing resizeSettling-gated forwarding loop, which also drops rather than replays suppressed frames, relying on the post-quiescence capture-pane snapshot (applyNegotiatedSize) to bring every subscriber back in sync instead. Frames that pass the resize gate are fed into the hub's single BatchWindow (Epic 2.1) rather than broadcast directly, so N attached subscribers share one accumulation/coalesce pass per burst instead of each running their own.

func (*StreamHub) RequestResize 

func (h *StreamHub) RequestResize(ctx context.Context, id SubscriberID, size TerminalSize)

RequestResize records subscriber id's vote for size and re-runs negotiation, but only if that subscriber was attached with SubscriberCapability.CanResize == true (Task 1.3.1c) — a read-only sink's vote is rejected and logged, never applied. Requesting resize for an unknown SubscriberID is a no-op.

ctx should be scoped to the calling connection's lifetime (canceled when that caller disconnects), not a fixed background timeout — it becomes the deadline applyNegotiatedSize bounds its SetWindowSize/CapturePaneContentRaw calls with, so an early disconnect stops those calls instead of always waiting out a fixed ceiling.

func (*StreamHub) SlowSubscriberDropsTotal 

func (h *StreamHub) SlowSubscriberDropsTotal() int64

SlowSubscriberDropsTotal returns the number of times a subscriber has been evicted for staying slow past its grace period — incremented exactly once per eviction, never once per dropped frame. See the field doc comment on StreamHub for scoping notes (Epic 3.2 owns wiring this to a real metrics backend).

func (*StreamHub) State 

func (h *StreamHub) State() HubLifecycleState

State returns the hub's current HubLifecycleState.

func (*StreamHub) SubscriberCount 

func (h *StreamHub) SubscriberCount() int

SubscriberCount returns the number of currently attached subscribers.

func (*StreamHub) TryFlush 

func (h *StreamHub) TryFlush()

TryFlush flushes the hub's BatchWindow immediately if anything is pending, with FlushOpportunistic as the reason. This is the hub-level hook a raw output feed (pumpControlModeOutputIntoHub) calls once it has drained every frame immediately available on its subscription channel — mirroring the `default: break coalesce` point in server/services/connectrpc_websocket.go's legacy per-connection coalesce loop, so a hub-owned burst is not forced to always pay MaxBatchWindow's full ceiling latency when the feed momentarily has nothing left to drain. A no-op if nothing is buffered (e.g. a resize is in progress and OnRawOutput has been suppressing every frame).

func (*StreamHub) TryStartPump added in v1.49.0 

func (h *StreamHub) TryStartPump() bool

TryStartPump atomically claims the right to run this hub's raw-output pump, returning true for exactly one caller. AttachSubscriber's doc comment notes that reactivating a HubTornDown hub is HubRegistry's policy to own — this is that policy's missing half: a hub that has fully torn down (0 subscribers, grace period expired) has no pump goroutine left feeding it (MarkPumpExited already flipped this back to false), and nothing else restarts one on reattach, silently losing live output for the rest of the process's life. Callers (HubRegistry.GetOrCreate) should call this unconditionally after every LoadOrCompute — a healthy hub with pumpActive already true simply loses the CAS and spawns nothing, so this is safe to call on every reconnect, not just a detected cache-hit case.

type StreamOwnershipLock 

type StreamOwnershipLock struct {
	// contains filtered or unexported fields
}

StreamOwnershipLock resolves STAPLER_SQUAD_USE_STREAM_HUB (plus any future per-session override, Story 3.3.1) into a StreamPath exactly once per tmux session, at the first connection's attach time, and sticks that resolution for the rest of the session's lifetime — see ADR-003. A per-connection re-read (Epic 2.2's useStreamHub placeholder) would let a flag flip mid-rollout split one session across two owners; this type closes that window by construction rather than by convention, the same safety property session/tmux.TmuxSession.controlModeStartMu gives the legacy 0->1 refcount transition.

Story 3.1.2 widens this into the shared mutual-exclusion primitive both legacy StartControlMode and hub creation acquire; today it only owns the resolve-once-and-cache behavior (Story 3.1.1).

func AcquireOwnershipLock 

func AcquireOwnershipLock(sessionName string) *StreamOwnershipLock

AcquireOwnershipLock returns the StreamOwnershipLock for sessionName, creating it if this is the first call for that name. Every caller passing the same sessionName gets the same *StreamOwnershipLock instance, which is what makes the mutual exclusion in Story 3.1.2 possible.

func (*StreamOwnershipLock) AcquireAndResolve 

func (l *StreamOwnershipLock) AcquireAndResolve(flagValue bool, fn func(StreamPath) error) error

AcquireAndResolve holds this lock's mutex for the full duration of fn, resolving flagValue into this session's sticky StreamPath first (identical semantics to Resolve) and passing it to fn before releasing. Unlike Resolve/ResolveExpecting — which only hold the mutex across the cheap resolve-and-cache step, then let the caller act on the result unsynchronized — AcquireAndResolve is Story 3.1.2's actual mutual-exclusion primitive: Instance.StartControlMode and HubRegistry.GetOrCreate both call it (not just Resolve) around their own side effect (starting the control-mode subprocess, or creating the hub), so a concurrent GetOrCreate genuinely blocks on an in-flight StartControlMode for the same session (and vice versa) instead of both racing to resolve first and only checking the outcome afterward. This closes the gap where mutual exclusion held only for callers that remembered to check ownership before proceeding — StartControlMode now enforces it unconditionally for every caller, present or future.

func (*StreamOwnershipLock) AcquireAndResolveExpecting 

func (l *StreamOwnershipLock) AcquireAndResolveExpecting(flagValue bool, want StreamPath, fn func() error) error

AcquireAndResolveExpecting is AcquireAndResolve plus ResolveExpecting's want-path assertion: fn only runs if the resolved path matches want: otherwise AcquireAndResolveExpecting returns ErrOwnershipResolvedToOtherPath without running fn, exactly like ResolveExpecting's error contract, but with the same held-lock guarantee AcquireAndResolve provides.

func (*StreamOwnershipLock) Resolve 

func (l *StreamOwnershipLock) Resolve(flagValue bool) StreamPath

Resolve turns flagValue into a StreamPath the first time it is called on this lock, caches that result, and returns the cached StreamPath on every subsequent call — regardless of what flagValue later callers pass. This is what makes a flag flip mid-rollout safe: the first connection's resolution sticks for the session's lifetime, so a second connection arriving after the flag changed in the environment still observes the original decision.

Story 3.3.1: before falling back to flagValue (the global default), Resolve consults sessionOverrideLookup for this lock's own session name. A recorded override forcing PathHubOwned wins regardless of flagValue — the per-session canary mechanism — but never overrides an already-sticky resolution, same as the global flag.

func (*StreamOwnershipLock) ResolveExpecting 

func (l *StreamOwnershipLock) ResolveExpecting(flagValue bool, want StreamPath) (StreamPath, error)

ResolveExpecting resolves the lock exactly like Resolve, but additionally asserts that the resolved StreamPath matches want — the caller's own intended role (hub creation expects PathHubOwned; legacy StartControlMode expects PathLegacyPerConnection). If another caller's resolution already won and it disagrees with want, ResolveExpecting returns the actual resolved path plus ErrOwnershipResolvedToOtherPath instead of proceeding, so HubRegistry.GetOrCreate and the legacy per-connection entry point (both in package server/services, which cannot itself hold this lock's internal state) can refuse to create a competing owner and instead join whichever path actually won (Story 3.1.2 / Task 3.1.2b).

type StreamPath 

type StreamPath int

StreamPath identifies which of the two mutually-exclusive ownership models a tmux session's stream is resolved to. Exactly two values exist; adding a third without updating every switch over StreamPath fails the `exhaustive` linter check. 

const (
	// PathLegacyPerConnection is the pre-existing model where each
	// connection owns its own resize/quiescence/capture pipeline.
	PathLegacyPerConnection StreamPath = iota

	// PathHubOwned is the new model where a single StreamHub owns
	// resize/quiescence/capture for all of a session's subscribers.
	PathHubOwned
)

type SubscriberCapability 

type SubscriberCapability struct {
	// CanResize allows the subscriber's RequestResize votes to count toward
	// NegotiatedSize. A read-only sink (e.g. a passive log tailer) sets this
	// false so it can never influence the pane's dimensions.
	CanResize bool

	// CanWrite allows the subscriber to send input to the session.
	CanWrite bool
}

SubscriberCapability describes what a Subscriber is permitted to do against the hub, independent of its Transport.

type SubscriberID 

type SubscriberID string

SubscriberID uniquely identifies one attached Subscriber. It is a newtype over string so a raw string can't be passed where a SubscriberID is expected without an explicit conversion.

func NewSubscriberID 

func NewSubscriberID() SubscriberID

NewSubscriberID returns a new, globally-unique SubscriberID.

type TerminalSize 

type TerminalSize struct {
	// contains filtered or unexported fields
}

TerminalSize is a validated {cols, rows} pane-dimension pair — the single shared representation used by RequestResize, ResizeVote, and NegotiatedSize instead of three independently-inlined shapes (or a bare (cols, rows int) pair a caller could silently transpose). Construct only via NewTerminalSize; the zero value is never returned by it and is not a valid size (per primitive-obsession-checklist.md and type-driven-design).

func NewTerminalSize 

func NewTerminalSize(cols, rows int) (TerminalSize, error)

NewTerminalSize validates that cols and rows are both positive and returns the resulting TerminalSize, or a non-nil error and the zero TerminalSize if either dimension is non-positive.

func (TerminalSize) Cols 

func (t TerminalSize) Cols() int

Cols returns the terminal's column count.

func (TerminalSize) Rows 

func (t TerminalSize) Rows() int

Rows returns the terminal's row count.

type Transport 

type Transport interface {
	// Send delivers a frame of output bytes to the subscriber. A non-nil
	// error is treated as a delivery failure and results in the subscriber
	// being evicted, the same as a slow/blocked subscriber.
	Send(data []byte) error

	// Close releases any resources held by the transport.
	Close() error
}

Transport is the delivery mechanism a Subscriber uses to receive hub output and be torn down. Implementations decouple the hub from any specific wire protocol (WebSocket, ssq-mux, in-memory test double).

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.