v1.3.0 Design: Broadcast Streams & Push-Source Scaling

Status: Design — approved for implementation Target version: v1.3.0 Author: Sumit Dhaka Scope: Manager+worker cluster mode only. Standalone mode is unaffected.

Problem Statement

TRAM v1.2.x has an architectural gap for HTTP push-source pipelines (webhook, prometheus_rw) in manager+worker mode:

  1. Manager dispatches a stream pipeline to one worker. WebhookSource registers an in-process queue on that worker. External systems have no stable, load-balanced entry point to reach it.
  2. Single-worker execution is a throughput bottleneck for high-volume sources (e.g. prometheus_rw at 1k–10k TPS).
  3. Stream pipelines never call _post_run_complete, so the manager has zero live metrics for running streams. Batch pipelines only report at completion — manager is blind while a batch is executing.
  4. Load-based dispatch uses active_runs count — a weak proxy that cannot distinguish a pipeline processing 100 TPS from one processing 1 TPS.
  5. No API tells operators which worker owns a pipeline or what its ingress endpoint is.
  6. The manager Deployment + RWO PVC causes downtime on every helm upgrade (strategy: Recreate).

Out of scope for v1.3.0

Concepts

placement_group_id

Every multi-worker dispatch creates a placement group — a logical unit representing one pipeline running across a set of workers simultaneously.

placement_group_id: "prom-ingest-20260417-ab12"   (pipeline_name + timestamp + short hash)
run_id per worker:  "prom-ingest-20260417-ab12-w0"
                    "prom-ingest-20260417-ab12-w1"
                    "prom-ingest-20260417-ab12-w2"

placement_group_id is the operator-facing identity for a multi-worker stream. The manager tracks placements by group, not individual run_ids.

Placement slot identity

Within a placement group, each worker slot has a stable identity:

slot_id = (placement_group_id, worker_index)

slot_id is the durable reconciliation key. run_id is the current execution instance for that slot and changes on re-dispatch after a worker restart:

slot 0 run_id:  "prom-ingest-20260417-ab12-w0"
slot 1 run_id:  "prom-ingest-20260417-ab12-w1"

# after re-dispatch to slot 1:
slot 1 run_id:  "prom-ingest-20260417-ab12-w1-r1"

Reconciliation matches by persisted slot metadata, not by requiring run_id to be immutable.

slot_id = (placement_group_id, worker_index) in all modes.

For count: all (v1.3.0), the worker assigned to slot i is fixed for the group lifetime — the slot’s worker_url never changes unless that specific worker is re-dispatched after a restart. Recovery re-dispatches the same worker to the same slot index.

For count: N (v1.3.1), the slot index is still the durable identity, but the worker_url assignment is mutable — on failover, slot i can be reassigned to any healthy spare. This separation (stable slot index, mutable worker assignment) is the reason count: N runtime is deferred: the reconciliation and slots_json model must be in place first.

Degraded placement

A dispatch that partially succeeds is degraded, not failed.

Slots accepted Status
All N running
1 to N-1 degraded — log WARNING; alerts fire if configured
0 error — reschedule per normal error policy

Degraded is a valid operational state, not an error. The placement API exposes it so operators can act on it.

A — workers: block (replaces dispatch: string)

Runtime scope

Feature v1.3.0 v1.3.1
count: all dispatch + reconciliation ✓ ✓
count: 1 (existing single-worker path) ✓ ✓
count: N (logical slot model, failover) schema only — runtime error if specified ✓
list: [worker_ids] placement schema only — runtime error if specified ✓

The schema is forward-compatible. count: N and list: are accepted by the parser and linter but rejected at dispatch time in v1.3.0 with a clear error message. This keeps the door open for v1.3.1 without changing YAML structure.

Standalone mode

In standalone mode there is no worker placement layer. workers: field is accepted and stored but never evaluated. Linter rules L006–L010 are all suppressed in standalone mode — they are meaningless without a worker pool.

Pipeline YAML

# HTTP push source — default is count: all when workers: is omitted
name: prom-ingest
schedule:
  type: stream
source:
  type: prometheus_rw
  path: prom-rw

# Kafka — explicit count (schema valid; v1.3.1 runtime)
name: kafka-ingest
schedule:
  type: stream
source:
  type: kafka
  topic: pm-data
  group_id: tram-pm-ingest
workers:
  count: 3   # accepted by schema; runtime error until v1.3.1

# Pin to specific workers (schema valid; v1.3.1 runtime)
workers:
  list:
    - tram-worker-0
    - tram-worker-1

workers: field semantics

workers: value v1.3.0 behaviour v1.3.1 behaviour
omitted, push HTTP source default → count: all same
omitted, all other sources default → count: 1 same
count: 1 least-loaded healthy worker (existing) same
count: all all healthy workers same
count: N (integer ≥ 2) runtime error: not yet supported N least-loaded workers
list: [w0, w1] runtime error: not yet supported exactly those workers

count and list are mutually exclusive — model validator enforces this. list accepts worker_id (pod name); manager resolves to URL internally.

Down-worker policy (v1.3.0 — count: all only)

Degraded slot filled by re-dispatch to the same worker when it recovers. count: all keeps all healthy workers in the group — a recovering worker re-joins automatically via PlacementReconciler. There is no “group is full” condition for count: all.

Data model — tram/models/pipeline.py

class WorkersConfig(BaseModel):
    count: int | Literal["all"] | None = None
    list: list[str] | None = None   # worker_ids (pod names); v1.3.1 runtime

    @model_validator(mode="after")
    def validate_workers(self) -> WorkersConfig:
        if self.count is not None and self.list is not None:
            raise ValueError("workers.count and workers.list are mutually exclusive")
        if isinstance(self.count, int) and self.count < 1:
            raise ValueError("workers.count must be >= 1")
        if self.list is not None:
            if len(self.list) == 0:
                raise ValueError("workers.list must not be empty")
            if len(self.list) != len(set(self.list)):
                raise ValueError("workers.list must not contain duplicate worker IDs")
        return self

class PipelineConfig(BaseModel):
    ...
    workers: WorkersConfig | None = None   # None → default applied at validation

    @model_validator(mode="after")
    def apply_workers_default(self) -> PipelineConfig:
        if self.workers is None:
            source_type = self.source.type if self.source else None
            if source_type in ("webhook", "prometheus_rw"):
                self.workers = WorkersConfig(count="all")
            else:
                self.workers = WorkersConfig(count=1)
        return self

Linter rules — tram/pipeline/linter.py

Linter receives tram_mode: str from caller (router/controller/CLI). tram validate reads TRAM_MODE from env or --mode flag. Rules L006–L010 are suppressed entirely in standalone mode.

Rule Condition Mode Severity
L006 HTTP push source (webhook, prometheus_rw) without count: all — LB routes to all workers; non-registered workers return 404, data is lost manager Error
L007 Poll/batch source (sftp, s3, rest, sql, local, ftp, gcs, azure_blob, gnmi, redis, influxdb, websocket, corba, elasticsearch, clickhouse) with any multi-worker spec (count > 1, count: all, or list:) — duplicate reads manager Error
L008 syslog or snmp_trap source — UDP broadcast not supported; planned v1.3.1 manager Error
L009 Queue source (kafka, nats, amqp) with integer count > configured worker pool size — will always degrade manager Warning
L010 Non-queue source with integer count > configured worker pool size manager Warning

L009 and L010 target different source classes to avoid double-warning on queue sources. A queue source that exceeds pool size triggers L009 only.

WorkerPool changes — tram/agent/worker_pool.py

New: worker_id → URL mapping (populated from health poll):

self._worker_ids: dict[str, str] = {}   # {worker_id: url}

# in _poll_all(), add:
if wid := data.get("worker_id"):
    self._worker_ids[wid] = url

New: resolve() — candidate selection (v1.3.0: count:1 and count:all only):

def resolve(self, cfg: WorkersConfig) -> list[str]:
    """Return ordered candidate URLs based on workers config.

    v1.3.0 supports count:1 and count:all only.
    count:N (integer ≥ 2) and list: raise NotImplementedError at dispatch time.
    """
    if cfg.list is not None:
        raise NotImplementedError("workers.list placement is not supported until v1.3.1")
    if isinstance(cfg.count, int) and cfg.count > 1:
        raise NotImplementedError("workers.count:N (N>1) placement is not supported until v1.3.1")
    healthy = sorted(
        [(url, self.load_score(url)) for url, h in self._health.items() if h["ok"]],
        key=lambda x: x[1]
    )
    n = len(healthy) if cfg.count == "all" else 1
    return [url for url, _ in healthy[:n]]

New: multi_dispatch() — single dispatch method:

def multi_dispatch(
    self,
    placement_group_id: str,
    pipeline_name: str,
    yaml_text: str,
    workers_cfg: WorkersConfig,
    callback_url: str = "",
) -> BroadcastResult:
    """Dispatch pipeline to worker candidates determined by workers_cfg.

    v1.3.0 supports count:1 and count:all.
    count:N and list: are schema-valid but raise NotImplementedError here.
    """

Existing dispatch() becomes a thin wrapper:

def dispatch(self, run_id, pipeline_name, yaml_text, schedule_type, callback_url=""):
    # kept for internal backwards compat; new callers use multi_dispatch
    result = self.multi_dispatch(run_id, pipeline_name, yaml_text,
                                 WorkersConfig(count=1), callback_url)
    return result.accepted[0] if result.accepted else None

BroadcastResult dataclass:

@dataclass
class BroadcastResult:
    placement_group_id: str
    accepted: list[str]   # worker URLs that returned 202
    rejected: list[str]   # worker URLs that failed or were unhealthy
    status: Literal["running", "degraded", "error"]

B — Worker Public Ingress Security

The problem

Mounting /webhooks on the worker app is fine in isolation. The problem arises when the worker port (:8766) is exposed via a K8s NodePort: /agent/* endpoints become externally reachable on the same port. /agent/run, /agent/stop, /agent/status allow arbitrary pipeline execution and control. Unacceptable.

Solution: split ingress surface

Port Name Routes Exposure
:8766 Agent port /agent/* Internal only (headless DNS)
:8767 Ingress port /webhooks/* NodePort for external push traffic 
# tram/agent/server.py
def create_worker_ingress_app() -> FastAPI:
    """Minimal push-traffic receiver. No /agent/* routes."""
    app = FastAPI(title="TRAM Worker Ingress", openapi_url=None)
    app.include_router(webhooks_router)
    if os.environ.get("TRAM_API_KEY"):
        app.add_middleware(APIKeyMiddleware)
    return app

APIKeyMiddleware import contract

tram/api/middleware.py is safe to import in worker mode — it depends only on tram/core/config.py, no manager-only modules.

This is a contract, not a coincidence. Enforced by:

Failure policy — coupled shutdown

Split ports exist for exposure isolation only, not independent availability:

Failure Without coupling Correct
Agent (:8766) crashes Ingress accepts traffic; no processing Worker exits
Ingress (:8767) crashes Manager sees healthy; push traffic dead Worker exits

Rule: if either listener thread exits, the worker process exits. Implementation: main thread joins both listener threads; either exits → os.kill(os.getpid(), SIGTERM) → both stop → K8s restarts pod.

Composite health: GET /agent/health returns ok: false when ingress thread is not alive. Readiness probe target unchanged (:8766), but handler checks ingress thread:

@app.get("/agent/health")
def health():
    ingress_alive = app.state.ingress_thread.is_alive()
    return {
        "ok": ingress_alive,
        "worker_id": worker_id,
        "active_runs": len(active),
        "ingress_up": ingress_alive,
    }

K8s removes pod from NodePort endpoints when readiness fails — no new push traffic reaches the restarting pod.

TRAM_WORKER_INGRESS_PORT — AppConfig

worker_ingress_port: int  # TRAM_WORKER_INGRESS_PORT, default 8767

Helm changes

Worker StatefulSet gains a second container port:

ports:
  - name: agent    # containerPort 8766 — internal only
  - name: ingress  # containerPort 8767 — NodePort candidate (v1.3.1 dynamic provisioning)

Readiness/liveness probes remain on agent port (:8766) using composite health endpoint.

C — Unified Pipeline Stats & Load-Aware Dispatch

The gap

Current state:

PipelineStats — tram/agent/metrics.py

Single stats object used for both stream and batch:

@dataclass
class PipelineStats:
    run_id: str
    pipeline_name: str
    schedule_type: str          # "stream" | "batch"
    # record counters (cumulative since run start)
    records_in: int = 0
    records_out: int = 0
    records_skipped: int = 0
    dlq_count: int = 0
    error_count: int = 0
    # byte counters (cumulative since run start)
    bytes_in: int = 0           # raw bytes read from source
    bytes_out: int = 0          # bytes written to all sinks combined
    # rolling error window (reset each heartbeat interval)
    errors_last_window: list[str] = field(default_factory=list)
    _lock: threading.Lock = field(default_factory=threading.Lock, compare=False)

    def increment(
        self, *,
        records_in: int = 0, records_out: int = 0, skipped: int = 0,
        dlq: int = 0, bytes_in: int = 0, bytes_out: int = 0,
        errors: list[str] | None = None,
    ) -> None:
        with self._lock:
            self.records_in      += records_in
            self.records_out     += records_out
            self.records_skipped += skipped
            self.dlq_count       += dlq
            self.bytes_in        += bytes_in
            self.bytes_out       += bytes_out
            if errors:
                self.error_count += len(errors)
                self.errors_last_window.extend(errors[-10:])

    def snapshot_and_reset_window(self) -> dict:
        with self._lock:
            snap = {
                "records_in":          self.records_in,
                "records_out":         self.records_out,
                "records_skipped":     self.records_skipped,
                "dlq_count":           self.dlq_count,
                "error_count":         self.error_count,
                "bytes_in":            self.bytes_in,
                "bytes_out":           self.bytes_out,
                "errors_last_window":  list(self.errors_last_window),
            }
            self.errors_last_window.clear()
        return snap

ActiveRun gains stats: PipelineStats for all run types (stream and batch). Replaces the earlier StreamMetrics design — one model for both.

Executor integration — tram/pipeline/executor.py

Both stream_run() and batch_run() accept stats: PipelineStats | None.

Where bytes are counted:

# source read loop (both modes)
for raw_bytes, meta in source.read():
    if stats:
        stats.increment(bytes_in=len(raw_bytes))
    ...

# sink write
serialised = serializer_out.serialize(records)
sink.write(serialised, meta)
if stats:
    stats.increment(bytes_out=len(serialised))

Worker stats reporting thread

A single reporting thread handles all active runs (stream and batch):

def _stats_loop(state: WorkerState, manager_url: str, interval: int) -> None:
    while not state.stats_stop.wait(interval):
        for run in state.snapshot():
            if run.stats is None:
                continue
            snap = run.stats.snapshot_and_reset_window()
            payload = {
                "worker_id":      state.worker_id,
                "pipeline_name":  run.pipeline_name,
                "run_id":         run.run_id,
                "schedule_type":  run.schedule_type,
                "uptime_seconds": (utcnow() - run.started_at_dt).total_seconds(),
                "timestamp":      utcnow().isoformat(),
                "is_final":       False,   # periodic; not the final report
                **snap,
            }
            _post_stats(f"{manager_url}/api/internal/pipeline-stats", payload)

Fire-and-forget. A missed interval means stale data, not a fatal event.

Batch runs additionally emit a final stats report with is_final: true immediately before calling _post_run_complete. This final report carries the cumulative bytes_in, bytes_out, records_in, records_out, and records_skipped totals for the entire run.

Final byte totals must reach run_history. The worker therefore includes the same totals in the RunCompletePayload it POSTs to /api/internal/run-complete. The is_final stats report is a load-management signal (remove from StatsStore); the RunCompletePayload is the durable record. Both are sent, and on_worker_run_complete writes bytes to run_history. Intermediate periodic reports are best-effort only and are never used as the authoritative source for final counts.

on_worker_run_complete also calls StatsStore.remove(run_id) as a belt-and-suspenders fallback for runs that crash without sending a final report.

Configuration

# AppConfig
stats_interval: int  # TRAM_STATS_INTERVAL, default 30
                     # (replaces TRAM_HEARTBEAT_INTERVAL — broader concept)

Manager stats receiver — tram/api/routers/internal.py

class PipelineStatsPayload(BaseModel):
    worker_id: str
    pipeline_name: str
    run_id: str
    schedule_type: str
    uptime_seconds: float
    timestamp: str
    records_in: int
    records_out: int
    records_skipped: int
    dlq_count: int
    error_count: int
    bytes_in: int
    bytes_out: int
    errors_last_window: list[str] = []
    is_final: bool = False   # True on last report for a batch run

@router.post("/api/internal/pipeline-stats")
async def pipeline_stats(payload: PipelineStatsPayload, request: Request) -> dict:
    store = request.app.state.stats_store
    if payload.is_final:
        store.remove(payload.run_id)   # batch run done — evict from load scoring
        # Final byte/record totals are carried by RunCompletePayload, not here.
        # This endpoint's only job on is_final is to remove the run from StatsStore.
    else:
        store.update(payload)
    return {"ok": True}

Endpoint is write-only. All reactive logic lives in PlacementReconciler. on_worker_run_complete also calls StatsStore.remove(run_id) as a fallback for any batch run that does not emit a final stats report (e.g. crash).

Authoritative final totals path: RunCompletePayload already carries records_in, records_out, records_skipped. It gains bytes_in: int = 0 and bytes_out: int = 0 so on_worker_run_complete can write byte totals to run_history. The is_final stats report is purely a load-management signal — it triggers StatsStore eviction but is not the source of truth for run_history.

StatsStore — tram/agent/stats_store.py

Replaces the earlier HeartbeatStore design. Keyed by run_id — the only identifier that is unique per active execution. pipeline_name and worker_id are secondary indices, derived from stored payloads.

Why run_id as primary key:

Batch run expiry — explicit remove-on-complete path:

Completed batch runs are removed from the store immediately when the run-complete callback is processed by the manager (or when the worker’s final stats report is tagged is_final: true). This prevents stale batch entries from inflating load_score() for up to 3 × interval after the run has already finished.

The is_final flag is set by the worker stats loop on the last report emitted before _post_run_complete. The manager’s /api/internal/pipeline-stats handler calls StatsStore.remove(run_id) when it sees is_final: true. on_worker_run_complete also calls StatsStore.remove(run_id) as a belt-and-suspenders fallback for any run that sends no final stats report.

class StatsStore:
    """In-memory store of latest PipelineStats, keyed by run_id.

    Primary key: run_id (unique per active execution).
    Secondary indices: pipeline_name → [run_id], worker_id → [run_id].

    Covers both stream and batch active runs.
    Not persisted — restart reconciliation handled by PlacementReconciler.

    Staleness contract
    ------------------
    Stale entries (age > 3 × interval) are RETAINED in the store so the
    reconciler can detect them. However, the load-scoring views EXCLUDE stale
    entries — a crashed or silent worker must not inflate load_score() until
    the staleness timeout expires.

    - get_by_run_id()  — stale-aware: returns the entry regardless of age.
                         Used by PlacementReconciler (stale detection) and
                         Placement API (per-slot stats lookup).
    - for_worker()     — excludes stale entries. Used by load_score() only.
    - for_pipeline()   — excludes stale entries. Used by aggregate/live views
                         (e.g. cluster/streams totals) only — NOT by the
                         placement detail API, which needs stale visibility.
    - all_active()     — excludes stale entries.

    Placement API data access
    -------------------------
    GET /api/pipelines/{name}/placement iterates slots from
    broadcast_placements.slots_json (source of truth for all slots including
    stale/degraded) and calls get_by_run_id(slot.current_run_id) per slot.
    Per-sec fields are zeroed for stale entries. for_pipeline() is NOT used
    for placement detail — it would silently drop stale slots.
    """

    def update(self, payload: PipelineStatsPayload) -> None:
        """Insert or replace entry for payload.run_id.""" ...

    def remove(self, run_id: str) -> None:
        """Explicit removal — called on run-complete or is_final stats.""" ...

    def get_by_run_id(self, run_id: str) -> PipelineStatsPayload | None:
        """Returns entry regardless of staleness — reconciler use only.""" ...

    def for_pipeline(self, pipeline_name: str) -> list[PipelineStatsPayload]:
        """Non-stale entries for this pipeline.""" ...

    def for_worker(self, worker_id: str) -> list[PipelineStatsPayload]:
        """Non-stale entries for this worker — used by load_score().""" ...

    def all_active(self) -> list[PipelineStatsPayload]:
        """All non-stale entries.""" ...

    def is_stale(self, entry: PipelineStatsPayload, interval: int) -> bool:
        """age > 3 × interval → stale.""" ...

PipelineStatsPayload gains is_final: bool = False — set by the worker on the last stats report for a batch run.

Load-aware dispatch — WorkerPool.load_score()

def load_score(self, worker_url: str) -> float:
    """Bytes/sec throughput across all active runs on this worker.

    Returns float for sorting. Lower = less loaded.
    Falls back to active_runs count (scaled) when no stats data yet —
    handles new workers that have not yet sent a stats report.

    Uses worker_id (from _worker_ids[worker_url]) as the StatsStore lookup key,
    since StatsStore.for_worker() is indexed by worker_id, not by URL.
    """
    worker_id = self._worker_ids.get(worker_url)
    stats = self._stats_store.for_worker(worker_id) if worker_id else []
    if not stats:
        # bootstrapping fallback: no stats yet → use active_runs × 1 MB proxy
        return float(self._health[worker_url]["active_runs"]) * 1_000_000
    interval = self._stats_interval
    return sum(
        (s.bytes_in + s.bytes_out) / max(s.uptime_seconds, interval)
        for s in stats
    )

WorkerPool receives a StatsStore reference at construction. resolve() calls load_score() for all healthy workers before sorting.

D — PlacementReconciler — tram/agent/reconciler.py

Dedicated manager-side background thread. Owns all reactive logic — stale detection, re-dispatch, reconciling-window timeout. The stats endpoint is a write path only.

Why dedicated thread (not heartbeat-driven): If all workers for a pipeline go silent simultaneously, a heartbeat-driven approach fires nothing. A periodic loop has bounded worst-case detection latency regardless of traffic.

PlacementReconciler runs every min(TRAM_STATS_INTERVAL, 10) seconds:

  for each active broadcast_placement:           # v1.3.0: count:all groups only
    for each slot (slot = worker position):
      last_stats = StatsStore.get_by_run_id(slot.current_run_id)
      if last_stats is None OR last_stats.timestamp age > 3 × interval:
        mark slot stale
      if slot stale AND slot.worker is healthy (from health poll):
        re-dispatch slot to same worker (count:all — position is fixed):
          new run_id = {pg_id}-w{i}-r{restart_count}
          POST /agent/run → update slot.current_run_id in slots_json + DB

    if group.status == reconciling AND group age > 2 × interval:
      evaluate matched slots → running | degraded | re-dispatch unmatched

v1.3.1 will extend this loop with:

PlacementReconciler holds refs to:

Wired into create_app() lifespan alongside StatsStore.

E — Manager Restart Reconciliation

Two-phase on startup

Phase 1 — DB rehydration (immediate): _boot_load() reads broadcast_placements table, seeds _broadcast_placements and _active_placement_group, sets pipeline status reconciling.

Phase 2 — stats reconciliation (passive, within 2 × interval): Incoming stats reports matched by run_id against persisted slot run_id_prefix. Matched slot → running. After 2 × interval:

Worker restart — manager-driven only

Workers must not autonomously recreate placement state after a full pod restart. Reason: manager is the placement authority; old in-process queues are gone; self-recovery risks duplicate listeners or state divergence.

Recovery path is always:

  1. Manager detects stale stats (after 3 × interval)
  2. PlacementReconciler marks slot stale
  3. Worker comes back up; readiness probe passes
  4. Reconciler re-dispatches that slot: POST /agent/run with new run_id = {pg_id}-w{i}-r{restart_count}
  5. Worker starts fresh stream instance from that assignment

DB schema — broadcast_placements table

CREATE TABLE broadcast_placements (
    placement_group_id  TEXT PRIMARY KEY,
    pipeline_name       TEXT NOT NULL,
    slots_json          TEXT NOT NULL,
    -- JSON array of slot objects:
    -- {
    --   "worker_index":    0,
    --   "worker_url":      "http://tram-worker-0...:8766",
    --   "worker_id":       "tram-worker-0",
    --   "run_id_prefix":   "prom-ingest-20260417-ab12-w0",   -- immutable; restart reconciliation
    --   "current_run_id":  "prom-ingest-20260417-ab12-w0-r1",-- mutable; updated on re-dispatch
    --   "status":          "running"                          -- running|stale|degraded
    -- }
    -- run_id_prefix: used by Phase-2 manager-restart reconciliation (prefix match on incoming stats)
    -- current_run_id: used by live stale-detection (StatsStore.get_by_run_id exact match)
    target_count        TEXT NOT NULL,   -- "1" | "3" | "all"
    started_at          TIMESTAMP NOT NULL,
    status              TEXT NOT NULL,   -- running|degraded|error|stopped|reconciling
    stopped_at          TIMESTAMP
);

F — Placement API

GET /api/pipelines/{name}/placement

{
  "name": "prom-ingest",
  "workers": {"count": "all"},
  "schedule_type": "stream",
  "placement_group_id": "prom-ingest-20260417-ab12",
  "group_status": "running",
  "placements": [
    {
      "worker_id": "tram-worker-0",
      "worker_url": "http://tram-worker-0.tram-worker.default.svc:8766",
      "run_id": "prom-ingest-20260417-ab12-w0",
      "status": "running",
      "started_at": "2026-04-17T06:00:00Z",
      "uptime_seconds": 14400,
      "stats": {
        "records_in": 1800000,
        "records_out": 1799200,
        "records_skipped": 42,
        "bytes_in": 900000000,
        "bytes_out": 720000000,
        "dlq_count": 11,
        "error_count": 53,
        "records_in_per_sec": 125.0,
        "bytes_in_per_sec": 62500.0,
        "errors_last_window": []
      },
      "push_endpoint": {
        "internal": "http://tram-worker-0.tram-worker.default.svc:8767/webhooks/prom-rw"
      }
    },
    {
      "worker_id": "tram-worker-1",
      "run_id": "prom-ingest-20260417-ab12-w1",
      "status": "stale",
      "uptime_seconds": 14400,
      "stats": {"records_in": 1750000, "bytes_in_per_sec": 0.0},
      "push_endpoint": {
        "internal": "http://tram-worker-1.tram-worker.default.svc:8767/webhooks/prom-rw"
      }
    }
  ]
}

records_in_per_sec = records_in / uptime_seconds (stable rolling average). bytes_in_per_sec = bytes_in / uptime_seconds. push_endpoint.internal assembled from headless DNS + TRAM_WORKER_INGRESS_PORT. No K8s API call required.

GET /api/cluster/streams

{
  "streams": [
    {
      "pipeline_name": "prom-ingest",
      "placement_group_id": "prom-ingest-20260417-ab12",
      "group_status": "running",
      "worker_count": 3,
      "degraded_count": 0,
      "stale_count": 0,
      "total_records_in_per_sec": 375.0,
      "total_bytes_in_per_sec": 187500.0,
      "total_dlq_count": 33,
      "last_stats_at": "2026-04-17T10:00:00Z"
    }
  ]
}

G — Manager StatefulSet

Why

Deployment + strategy: Recreate + RWO PVC = guaranteed downtime on every helm upgrade. StatefulSet + volumeClaimTemplates ties PVC lifecycle to pod ordinal — rolling update works correctly. One manager pod still restarts during upgrade — better, not zero-downtime. True HA (standby manager) remains a backlog item.

Changes

Remove: helm/templates/manager-deployment.yaml

Add: helm/templates/manager-statefulset.yaml

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: -manager
spec:
  replicas: 1
  serviceName: -manager
  podManagementPolicy: OrderedReady
  updateStrategy:
    type: RollingUpdate   # replaces strategy.type: Recreate
  template:
    # pod spec identical to current Deployment
  volumeClaimTemplates:
    - metadata:
        name: manager-data
      spec:
        accessModes: ["ReadWriteOnce"]
        resources:
          requests:
            storage: 
        storageClassName:

Add: helm/templates/manager-headless-service.yaml — headless DNS for tram-manager-0.

Remove: manager-data- PVC from pvc.yaml (now owned by volumeClaimTemplates).

Existing ClusterIP service (service.yaml) unchanged — workers keep TRAM_MANAGER_URL = http://tram:8765.

Migration: manager.persistence.existingClaim in values.yaml. If provided, volumeClaimTemplates is skipped and the named PVC mounted directly. Users upgrading from v1.2.x set this to their existing manager-data-<release> PVC name.

Lifecycle Semantics

Broadcast / multi-worker start

  1. Manager generates placement_group_id
  2. WorkerPool.resolve(workers_cfg) selects candidate URLs by load score
  3. multi_dispatch() POSTs /agent/run to each candidate with run_id = {pg_id}-w{i}; records accepted/rejected
  4. BroadcastResult persisted to broadcast_placements table
  5. Pipeline status set: running degraded error

Stop (manual or delete)

  1. Manager iterates accepted worker list from BroadcastResult
  2. POSTs /agent/stop with matching run_id to each
  3. Workers signal stop_event; streams exit
  4. broadcast_placements.status = stopped; in-memory state cleared

Degraded placement

Manager restart

  1. _boot_load() seeds from broadcast_placements → reconciling status
  2. Stats reports matched by run_id prefix → slots resolved
  3. After 2 × interval: matched → running; partial → degraded; unmatched → re-dispatch

Worker restart

  1. Stats go stale (age > 3 × interval)
  2. PlacementReconciler marks slot stale
  3. Worker recovers; readiness probe passes on both ports
  4. Reconciler re-dispatches slot; worker starts fresh from new /agent/run
  5. Traffic resumes after K8s readiness gate passes

New API Summary

Method Path Description
GET /api/pipelines/{name}/placement Placement detail for one pipeline
GET /api/cluster/streams All active multi-worker stream placements
POST /api/internal/pipeline-stats Worker → Manager unified stats report

Environment Variables

Variable Default Description
TRAM_STATS_INTERVAL 30 Seconds between worker stats reports
TRAM_WORKER_INGRESS_PORT 8767 Worker ingress listener port

Implementation Checklist

A — workers: block & multi-dispatch

B — Worker Public Ingress Security

C — Unified Pipeline Stats

D — PlacementReconciler

E — Manager Restart Reconciliation

F — Placement API

G — Manager StatefulSet

H — Alert cooldown fix (issue #3)