Skip to content

Supervise pipeline delivery and persist failure context - #29

Merged
tmosleyIII merged 3 commits into
mainfrom
feat/supervised-pipeline-delivery
Oct 12, 2026
Merged

tmosleyIII merged 3 commits into
mainfrom
feat/supervised-pipeline-delivery

Conversation

@tmosleyIII

Copy link
Copy Markdown
Contributor

Summary

  • Supervise source, processor, and sink forwarding under one terminal result; fail and cancel the run on delivery errors, and apply backpressure without dropping events.
  • Wire source-to-sink pipelines directly; require spec.registrationOnly: true for intentional monitoring-only pipelines.
  • Persist a structured, credential-free first failure in the existing PipelineRun.error field and mark the implicated component unhealthy.
  • Add forwarding, control-plane, and process-driver integration tests, and document the remaining boundedness limitation.

Verification

  • go test ./...
  • go vet ./...
  • go build ./...
  • Stellar process-driver smoke: RUN_TIMEOUT=45s scripts/flowctl-process-e2e.sh (101 sink events, completed).

Remaining limitation

Bounded source completion is currently inferred from END_LEDGER; other finite source types need an explicit boundedness setting.

@chatgpt-codex-connector

chatgpt-codex-connector Bot commented Oct 12, 2026 •

Copy link
Copy Markdown

Codex Review Summary

This comment shows the latest Codex review activity on this pull request.

Review Status Commit Review trigger
📝 Code Review ✅ Completed 2026-10-12T13:03:19.046547Z 620b22d PR opened
ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

  • Open a pull request for review
  • Mark a draft as ready
  • Comment "@codex review" or "@codex security review".

Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings.

@chatgpt-codex-connector chatgpt-codex-connector Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 620b22ddfb

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

  • Open a pull request for review
  • Mark a draft as ready
  • Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

Comment on lines +141 to +142
for _, sink := range s.pipeline.Spec.Sinks {
info, err := s.getComponentEndpoint(sink.ID)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Respect declared inputs when wiring direct sinks

For a processor-free pipeline with multiple sources, this loop opens every sink for every source without checking sink.Inputs. Consequently, a sink declared with inputs: [source-a] also receives events from source-b whenever its accepted event types overlap or are unspecified, causing cross-delivery and potentially corrupting downstream results. Filter the direct source-to-sink edges using the configured input IDs.

Useful? React with 👍 / 👎.

Comment thread internal/runner/pipeline_runner.go Outdated
Comment on lines +333 to +336
streamOrch.cancel()
select {
case <-terminal:
case <-time.After(10 * time.Second):

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve a delivery failure returned while cancelling

When an operator stop or signal races with an edge failure whose Wait result has not yet reached terminal, this branch cancels the orchestrator and then discards the value received from terminal. If the errgroup had already captured a real delivery error, deliveryErr remains nil, so the run is recorded as STOPPED and the CLI exits successfully instead of reporting the failure. Capture and classify the result here just as in the preceding receive branch.

Useful? React with 👍 / 👎.

@tmosleyIII

Copy link
Copy Markdown
Contributor Author

Addressed both Codex findings in b68e3e3:

  • Direct source-to-sink wiring now honors each sink’s inputs; sources with no declared sink fail wiring instead of silently discarding events. Added a two-source/two-sink process-driver routing test.
  • On stop, the runner retains and classifies the terminal result even when it arrives after cancellation starts. Added tests for queued failures, failures arriving during cancellation, and normal operator cancellation.

Verified with go test ./..., go vet ./..., and go build ./....

@tmosleyIII
tmosleyIII merged commit 5f2f1e9 into main Oct 12, 2026
3 checks passed
@tmosleyIII
tmosleyIII deleted the feat/supervised-pipeline-delivery branch October 12, 2026 13:11
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant