fix: prevent double close on pipeline stop

[wolf31o2]

Summary by cubic

Make pipeline.Stop safe by closing internal channels only once, preventing panics from double close during repeated or concurrent stops. Added closeOnce to guard closing errorChan, filterChan, and outputChan.

^{Written for commit d675772. Summary will update automatically on new commits.}

Summary by CodeRabbit

• Bug Fixes

  • Pipeline shutdown operations are now atomic and race-free. The stop mechanism is safe to call multiple times and from concurrent execution contexts.

• Tests

  • Added unit tests to verify correct behavior during pipeline shutdown, including scenarios with plugin failures and idempotent stop operations.

_{✏️ Tip: You can customize this high-level summary in your review settings.}

[coderabbitai]

📝 Walkthrough

Walkthrough

The changes add synchronization mechanisms to the Pipeline struct to prevent multiple closes on internal channels. A new closeOnce field using sync.Once is introduced to guard channel closure operations. The Stop() method now uses closeOnce.Do() to ensure errorChan, filterChan, and outputChan are closed exactly once. Corresponding unit tests are added to verify Stop behavior, including handling plugin panics and verifying idempotent Stop calls.

Estimated code review effort

🎯 2 (Simple) | ⏱️ ~12 minutes

• Verify that the sync.Once pattern correctly prevents race conditions on channel closes
• Confirm panic propagation from plugin Stop() calls works as expected in TestStopWithPluginPanic
• Ensure TestStopIdempotent adequately exercises the idempotent behavior across multiple Stop() invocations
• Check that all three channels (errorChan, filterChan, outputChan) are properly closed within the closeOnce.Do() block

Pre-merge checks and finishing touches

✅ Passed checks (2 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately summarizes the main change: adding protection against double-closing channels in pipeline.Stop() via a sync.Once guard.

✨ Finishing touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch fix/stop-double-close

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[cubic-dev-ai]

No issues found across 1 file

[coderabbitai]

Actionable comments posted: 0

🧹 Nitpick comments (1)

pipeline/pipeline.go (1)

36-36: The closeOnce guard appears redundant given the existing stopOnce.

Since closeOnce.Do() is nested inside stopOnce.Do() (line 169 within line 138), and stopOnce already guarantees the entire Stop body executes exactly once, the additional closeOnce provides no extra protection. The channels are already guarded from double-close by stopOnce alone.

This may be defensive programming for future refactoring, but currently it adds unnecessary complexity.

Consider simplifying by removing closeOnce and relying on stopOnce:

-	closeOnce  sync.Once // protects channel closes

And in Stop():

-		// Close pipeline-owned channels safely
-		p.closeOnce.Do(func() {
-			close(p.errorChan)
-			close(p.filterChan)
-			close(p.outputChan)
-		})
+		// Close pipeline-owned channels
+		close(p.errorChan)
+		close(p.filterChan)
+		close(p.outputChan)

📜 Review details

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 22f50b0 and d675772.

📒 Files selected for processing (2)

• pipeline/pipeline.go (2 hunks)
• pipeline/pipeline_test.go (1 hunks)

🧰 Additional context used

🧬 Code graph analysis (1)

pipeline/pipeline_test.go (1)

pipeline/pipeline.go (1)

• New (39-47)

🔇 Additional comments (2)

pipeline/pipeline_test.go (2)

28-40: Good test coverage for panic propagation.

This test correctly verifies that Stop() propagates panics from plugin.Stop() rather than suppressing them, documenting important behavior for pipeline shutdown.

---

42-53: Excellent test for idempotent Stop() behavior.

This test validates the PR's objective of making Stop() safe to call multiple times, ensuring no errors or panics occur on repeated calls.

[wolf31o2]

Closing this in favor of #554