work-unit: control plane — output lifecycle status wiring #1416

Activity

1. strawgate

   added

   enhancementNew feature or requestNew feature or request

   productionRequired for production readinessRequired for production readiness

   P2: mediumMinor bug, missing validation, metric inaccuracyMinor bug, missing validation, metric inaccuracy

   work-unitScheduling ticket for one agent runScheduling ticket for one agent run

   architectureSystemic design improvements that eliminate classes of bugsSystemic design improvements that eliminate classes of bugs

   on Apr 6, 2026
2. github-actions commented on Apr 6, 2026

   

   github-actionsBot

   on Apr 6, 2026 – with GitHub Actions

   Contributor

   More actions

   tl;dr: Outcome A — Ready for implementation. The core blocker is already concrete: output lifecycle/readiness cannot be wired correctly because worker-pool acks collapse sink outcomes to success: bool, and readiness is currently independent of sink health.

   Recommendation

   Implement this issue as a focused protocol+diagnostics wiring change in worker_pool + pipeline + diagnostics + output tests: carry sink/lifecycle state through AckItem, update per-sink diagnostics, and gate readiness by explicit sink-health policy.

   Findings

   1. Ack protocol is too lossy for lifecycle truth

      • crates/logfwd/src/worker_pool.rs:82-87

        pub struct AckItem { ... pub success: bool, ... }

      • No sink identity or lifecycle stage (startup/degraded/retrying/failed/stopping) is propagated.

   2. Pipeline currently records output failure against pipeline name, not sink name

      • crates/logfwd/src/pipeline.rs:979

        self.metrics.output_error(self.name.as_str());

      • But outputs are keyed by configured output names:
        • crates/logfwd/src/pipeline.rs:331-337, 342-349 (metrics.add_output(&output_name, ...))
      • And output_error matches by output name only:
        • crates/logfwd-io/src/diagnostics.rs:197-203.

   3. Readiness does not use sink health at all

      • crates/logfwd-io/src/diagnostics.rs:553-555

        let ready = !self.pipelines.is_empty();

      • This is explicitly documented as not data-flow/sink-health based (545-552).

   4. Known in-code limitation confirms this gap

      • crates/logfwd/src/pipeline.rs:2382-2383

        // Per-sink error tracking ... not currently supported by the pool's AckItem protocol

   Verification

   Ran targeted existing readiness tests:

   cargo test -p logfwd-io ready_endpoint -- --nocapture

   Output:

   • test_ready_endpoint_no_pipelines_returns_503 ... ok
   • test_ready_endpoint_with_pipeline_returns_200 ... ok

   This confirms current /ready behavior is purely pipeline-registration based.

   Detailed Action Plan (mechanical)

   1. Extend worker ack semantics to carry sink lifecycle/status

      • File: crates/logfwd/src/worker_pool.rs (AckItem at 82-105, process_item at 565+, worker_task at 459+)
      • Exact changes:
        1. Add fields to AckItem:
           • sink_name: String (or interned/static variant)
           • outcome: OutputOutcome (new enum)
           • optional retry metadata (attempts, last_retry_after_ms)
        2. Replace (bool, send_latency_ns) return from process_item with structured outcome (OutputOutcome) + latency.
        3. Map send paths explicitly:
           • SendResult::Ok → Delivered
           • RetryAfter/... with eventual success/failure → RetriedThenDelivered / RetryExhausted
           • Rejected → PermanentFailure
           • timeout/panic/cancel/drain force-stop → FatalFailure / StoppingAborted

   2. Fix pipeline wiring to per-sink metrics and lifecycle

      • File: crates/logfwd/src/pipeline.rs (apply_pool_ack at 965-982)
      • Exact changes:
        1. Replace output_error(self.name.as_str()) with sink-name-aware update from AckItem.
        2. Update ack handling to record degraded/retrying/failure counters/state (not only success/fail).
        3. Preserve existing checkpoint semantics (ack_all_tickets) while enriching diagnostics state.

   3. Add sink lifecycle/readiness model to diagnostics

      • File: crates/logfwd-io/src/diagnostics.rs
      • Exact changes:
        1. Extend PipelineMetrics with sink-status storage (per output): starting|ready|degraded|failed|stopping + timestamps.
        2. Extend /api/pipelines payload to expose per-output lifecycle + retry pressure counters.
        3. Update /ready (serve_ready at 553-568) to evaluate configured sink-health policy instead of only !pipelines.is_empty().

   4. Introduce configurable readiness policy (minimal explicit policy)

      • Files:
        • crates/logfwd-config/src/... (server/output readiness config model)
        • crates/logfwd/src/main.rs and/or pipeline bootstrap wiring
      • Exact changes:
        1. Add policy enum (example):
           • AnySinkReady (lenient)
           • AllSinksReady (strict)
           • optional AllowDegradedFor: Duration
        2. Thread policy into diagnostics server readiness evaluation.

   5. Tests to write (representative and mechanical)

      • crates/logfwd/src/worker_pool.rs tests module
        • pattern: existing async worker_pool unit tests
        • add assertions for startup → retry/degraded → fatal → stopping transitions via new AckItem.outcome.
      • crates/logfwd/src/pipeline.rs tests (near test_async_fanout_output_errors_only_increment_failing_output)
        • assert per-sink error/lifecycle attribution is correct (not pipeline-name fallback).
      • crates/logfwd-io/src/diagnostics.rs tests (near ready endpoint tests at 1634+)
        • add readiness policy tests:
          • no healthy sinks => 503
          • degraded-within-policy => 200 (if configured)
          • failed sink under strict policy => 503

   6. Verification commands

      • cargo test -p logfwd-io diagnostics::tests::test_ready_endpoint_no_pipelines_returns_503 diagnostics::tests::test_ready_endpoint_with_pipeline_returns_200
      • cargo test -p logfwd --lib worker_pool
      • cargo test -p logfwd --lib pipeline
      • final gate: just ci

   Related Items

   Type	Link	Relevance
   Issue	#1416	This work-unit
   Issue	#1306	Shared control-plane substrate context (from issue scope)
   Issue	#1312	Output/runtime status architecture context (from issue scope)
   Issue	#1313	File-appender lifecycle integration context (from issue scope)
   Issue	#1315	Parent execution slice
   Issue	#1357	Adjacent sink-result semantics scope boundary
   PR	#701	Introduced async worker pool + async sinks foundation
   PR	#745	Follow-up fixes; includes retry/ack behavior refinements
   File	crates/logfwd/src/worker_pool.rs:82	Ack protocol currently too coarse (success: bool)
   File	crates/logfwd/src/pipeline.rs:979	Error attribution currently uses pipeline name
   File	crates/logfwd-io/src/diagnostics.rs:553	/ready currently ignores sink health

   Note

   🔒 Integrity filtering filtered 7 items

   Integrity filtering activated and filtered the following items during workflow execution.
   This happens when a tool call accesses a resource that does not meet the required integrity or secrecy level of the workflow.

   • issue:work-unit: control plane — output lifecycle status wiring #1416 (issue_read: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".)
   • issue:meta: I/O control-plane substrate — lifecycle, health, readiness, and status #1306 (issue_read: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".)
   • issue:I/O correctness refactor — correct by construction, not by patch #1315 (issue_read: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".)
   • issue:work-unit: logfwd-output — sink error classification and retry semantics #1357 (issue_read: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".)
   • issue:File appender output — new component with rotation, cleanup, path templates #1313 (issue_read: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".)
   • issue:Output delivery contract — retry, reject, checkpoint, fanout #1312 (issue_read: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".)
   • resource:search_pull_requests (search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".)

   ---

   What is this? | From workflow: Issue Triage

   Give us feedback! React with 🚀 if perfect, 👍 if helpful, 👎 if not.
3. strawgate commented on Apr 6, 2026

   

   strawgate

   on Apr 6, 2026

   OwnerAuthor

   More actions

   The first pure-seam extraction batch under this work-unit is now #1439 (logfwd-output sink and fanout health policy).

   That gives us a merge-safe local slice before the broader sink lifecycle wiring work lands.
4. strawgate

   added

   component:output/commonCross-cutting output behavior spanning multiple sinksCross-cutting output behavior spanning multiple sinks

   component:pipelinePipeline orchestration, batching, runtime flow, and worker coordinationPipeline orchestration, batching, runtime flow, and worker coordination

   on Apr 6, 2026
5. strawgate

   changed the title [-]work-unit: logfwd-output + worker_pool — sink lifecycle, degradation, and failure wiring[/-] [+]work-unit: control plane — output lifecycle status wiring[/+] on Apr 6, 2026
6. coderabbitai

   added

   component:diagnosticsDiagnostics endpoints, dashboard, health, probes, and trace explorerDiagnostics endpoints, dashboard, health, probes, and trace explorer

   on Apr 6, 2026
7. strawgate commented on Apr 7, 2026

   

   strawgate

   on Apr 7, 2026

   OwnerAuthor

   More actions

   Resolved by PRs #1454, #1480, #1505. Verified: OutputHealthEvent reducer, per-worker health tracking.
8. strawgate

   closed this as completedon Apr 7, 2026