Skip to content

fix: ensure plugins cannot close channels #554

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wolf31o2 merged 1 commit into from
Dec 15, 2025
Merged

fix: ensure plugins cannot close channels #554

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 4 additions & 6 deletions filter/chainsync/chainsync.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ import (
)

type ChainSync struct {
errorChan chan error
errorChan chan<- error
inputChan chan event.Event
outputChan chan event.Event
logger plugin.Logger
Expand All @@ -39,7 +39,6 @@ type ChainSync struct {
// New returns a new ChainSync object with the specified options applied
func New(options ...ChainSyncOptionFunc) *ChainSync {
c := &ChainSync{
errorChan: make(chan error),
inputChan: make(chan event.Event, 10),
outputChan: make(chan event.Event, 10),
}
Expand Down Expand Up @@ -332,13 +331,12 @@ func (c *ChainSync) Start() error {
func (c *ChainSync) Stop() error {
close(c.inputChan)
close(c.outputChan)
close(c.errorChan)
return nil
}

// ErrorChan returns the filter error channel
func (c *ChainSync) ErrorChan() chan error {
return c.errorChan
// SetErrorChan sets the error channel
func (c *ChainSync) SetErrorChan(ch chan<- error) {
c.errorChan = ch
}

// InputChan returns the input event channel
Expand Down
35 changes: 3 additions & 32 deletions filter/chainsync/chainsync_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,9 +30,6 @@ import (
"github.com/blinklabs-io/adder/event"
)

// MockLogger is a mock implementation of the plugin.Logger interface
type MockLogger struct{}

// MockAddress is a mock implementation of the ledger.Address interface
type MockAddress struct {
common.Address // Embed the common.Address struct
Expand Down Expand Up @@ -86,7 +83,7 @@ func (m MockAddress) Type() uint8 {
return 0
}

func (m *MockAddress) UnmarshalCBOR(data []byte) error {
func (m *MockAddress) UnmarshalCBOR(_ []byte) error {
return nil
}

Expand Down Expand Up @@ -136,15 +133,9 @@ func (m MockOutput) ToPlutusData() data.PlutusData {
}

func (m MockOutput) String() string {
return ""
return "mockOutput"
}

func (l *MockLogger) Info(msg string, args ...any) {}
func (l *MockLogger) Error(msg string, args ...any) {}
func (l *MockLogger) Debug(msg string, args ...any) {}
func (l *MockLogger) Warn(msg string, args ...any) {}
func (l *MockLogger) Trace(msg string, args ...any) {}

func TestNewChainSync(t *testing.T) {
c := New()
if c == nil {
Expand Down Expand Up @@ -178,18 +169,6 @@ func TestChainSync_Stop(t *testing.T) {
default:
t.Fatalf("expected outputChan to be closed")
}
select {
case <-c.errorChan:
default:
t.Fatalf("expected errorChan to be closed")
}
}

func TestChainSync_ErrorChan(t *testing.T) {
c := New()
if c.ErrorChan() == nil {
t.Fatalf("expected non-nil errorChan")
}
}

func TestChainSync_InputChan(t *testing.T) {
Expand Down Expand Up @@ -233,15 +212,7 @@ func mockStakeCredentialValue(
}
}

func mockStakeCredentialPtr(
credType uint,
hashBytes []byte,
) *common.Credential {
cred := mockStakeCredentialValue(credType, hashBytes)
return &cred
}

func mockAddress(addrStr string) common.Address {
func mockAddress(_ string) common.Address {
return common.Address{}
}

Expand Down
6 changes: 1 addition & 5 deletions filter/chainsync/plugin_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,8 +30,7 @@ func TestPluginRegistration(t *testing.T) {
assert.NotNil(t, p, "Plugin should be registered")

// Verify that the plugin implements the Plugin interface
_, ok := p.(plugin.Plugin)
assert.True(t, ok, "Plugin should implement the Plugin interface")
assert.NotNil(t, p, "Plugin should implement the Plugin interface")
}

func TestPluginStartStop(t *testing.T) {
Expand All @@ -51,9 +50,6 @@ func TestPluginChannels(t *testing.T) {
// Create a new plugin instance
p := NewFromCmdlineOptions()

// Verify that the error channel is not nil
assert.NotNil(t, p.ErrorChan(), "Error channel should not be nil")

// Verify that the input channel is not nil
assert.NotNil(t, p.InputChan(), "Input channel should not be nil")

Expand Down
10 changes: 4 additions & 6 deletions filter/event/event.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ import (
)

type Event struct {
errorChan chan error
errorChan chan<- error
inputChan chan event.Event
outputChan chan event.Event
logger plugin.Logger
Expand All @@ -32,7 +32,6 @@ type Event struct {
// New returns a new Event object with the specified options applied
func New(options ...EventOptionFunc) *Event {
e := &Event{
errorChan: make(chan error),
inputChan: make(chan event.Event, 10),
outputChan: make(chan event.Event, 10),
}
Expand Down Expand Up @@ -69,13 +68,12 @@ func (e *Event) Start() error {
func (e *Event) Stop() error {
close(e.inputChan)
close(e.outputChan)
close(e.errorChan)
return nil
}

// ErrorChan returns the filter error channel
func (e *Event) ErrorChan() chan error {
return e.errorChan
// SetErrorChan sets the error channel
func (e *Event) SetErrorChan(ch chan<- error) {
e.errorChan = ch
}

// InputChan returns the input event channel
Expand Down
19 changes: 12 additions & 7 deletions input/chainsync/chainsync.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,7 @@ type ChainSync struct {
kupoClient *kugo.Client
oConn *ouroboros.Connection
eventChan chan event.Event
errorChan chan error
errorChan chan<- error
status *ChainSyncStatus
dialFamily string
kupoUrl string
Expand Down Expand Up @@ -128,7 +128,6 @@ type StatusUpdateFunc func(ChainSyncStatus)
// New returns a new ChainSync object with the specified options applied
func New(options ...ChainSyncOptionFunc) *ChainSync {
c := &ChainSync{
errorChan: make(chan error),
eventChan: make(chan event.Event, 10),
intersectPoints: []ocommon.Point{},
status: &ChainSyncStatus{},
Expand Down Expand Up @@ -169,13 +168,12 @@ func (c *ChainSync) Start() error {
func (c *ChainSync) Stop() error {
err := c.oConn.Close()
close(c.eventChan)
close(c.errorChan)
return err
}

// ErrorChan returns the input error channel
func (c *ChainSync) ErrorChan() chan error {
return c.errorChan
// SetErrorChan sets the error channel
func (c *ChainSync) SetErrorChan(ch chan<- error) {
c.errorChan = ch
}

// InputChan always returns nil
Expand Down Expand Up @@ -315,7 +313,14 @@ func (c *ChainSync) setupConnection() error {
}
} else {
// Pass error through our own error channel
c.errorChan <- err
if c.errorChan != nil {
c.errorChan <- err
} else if c.logger != nil {
c.logger.Warn(fmt.Sprintf(
"error occurred but no error channel set: %s",
err,
))
}
}
}
}()
Expand Down
14 changes: 7 additions & 7 deletions output/embedded/embedded.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,15 +23,14 @@ import (
type CallbackFunc func(event.Event) error

type EmbeddedOutput struct {
errorChan chan error
errorChan chan<- error
eventChan chan event.Event
callbackFunc CallbackFunc
outputChan chan event.Event
}

func New(options ...EmbeddedOptionFunc) *EmbeddedOutput {
e := &EmbeddedOutput{
errorChan: make(chan error),
eventChan: make(chan event.Event, 10),
}
for _, option := range options {
Expand All @@ -51,7 +50,9 @@ func (e *EmbeddedOutput) Start() error {
}
if e.callbackFunc != nil {
if err := e.callbackFunc(evt); err != nil {
e.errorChan <- fmt.Errorf("callback function error: %w", err)
if e.errorChan != nil {
e.errorChan <- fmt.Errorf("callback function error: %w", err)
}
return
}
}
Expand All @@ -66,16 +67,15 @@ func (e *EmbeddedOutput) Start() error {
// Stop the embedded output
func (e *EmbeddedOutput) Stop() error {
close(e.eventChan)
close(e.errorChan)
if e.outputChan != nil {
close(e.outputChan)
}
return nil
}

// ErrorChan returns the input error channel
func (e *EmbeddedOutput) ErrorChan() chan error {
return e.errorChan
// SetErrorChan sets the error channel
func (e *EmbeddedOutput) SetErrorChan(ch chan<- error) {
Copy link
Copy Markdown

@cubic-dev-ai cubic-dev-ai bot Dec 6, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

P1: The errorChan is no longer initialized in New(), but Start() sends to it without a nil check. If SetErrorChan() is never called and a callback error occurs, sending to a nil channel will block the goroutine forever. Consider adding a nil check before sending, or documenting that SetErrorChan() must be called before Start().

Prompt for AI agents 
Check if this issue is valid — if so, understand the root cause and fix it. At output/embedded/embedded.go, line 75:

<comment>The `errorChan` is no longer initialized in `New()`, but `Start()` sends to it without a nil check. If `SetErrorChan()` is never called and a callback error occurs, sending to a nil channel will block the goroutine forever. Consider adding a nil check before sending, or documenting that `SetErrorChan()` must be called before `Start()`.</comment>

<file context>
@@ -66,16 +65,15 @@ func (e *EmbeddedOutput) Start() error {
-func (e *EmbeddedOutput) ErrorChan() chan error {
-	return e.errorChan
+// SetErrorChan sets the error channel
+func (e *EmbeddedOutput) SetErrorChan(ch chan&lt;- error) {
+	e.errorChan = ch
 }
</file context>
Fix with Cubic

e.errorChan = ch
}

// InputChan returns the input event channel
Expand Down
10 changes: 4 additions & 6 deletions output/log/log.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ import (
)

type LogOutput struct {
errorChan chan error
errorChan chan<- error
eventChan chan event.Event
logger plugin.Logger
outputLogger *slog.Logger
Expand All @@ -33,7 +33,6 @@ type LogOutput struct {

func New(options ...LogOptionFunc) *LogOutput {
l := &LogOutput{
errorChan: make(chan error),
eventChan: make(chan event.Event, 10),
level: "info",
}
Expand Down Expand Up @@ -81,13 +80,12 @@ func (l *LogOutput) Start() error {
// Stop the log output
func (l *LogOutput) Stop() error {
close(l.eventChan)
close(l.errorChan)
return nil
}

// ErrorChan returns the input error channel
func (l *LogOutput) ErrorChan() chan error {
return l.errorChan
// SetErrorChan sets the error channel
func (l *LogOutput) SetErrorChan(ch chan<- error) {
l.errorChan = ch
}

// InputChan returns the input event channel
Expand Down
10 changes: 4 additions & 6 deletions output/notify/notify.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,15 +28,14 @@ import (
var icon []byte

type NotifyOutput struct {
errorChan chan error
errorChan chan<- error
eventChan chan event.Event
logger plugin.Logger
title string
}

func New(options ...NotifyOptionFunc) *NotifyOutput {
n := &NotifyOutput{
errorChan: make(chan error),
eventChan: make(chan event.Event, 10),
title: "Adder",
}
Expand Down Expand Up @@ -166,13 +165,12 @@ func (n *NotifyOutput) Start() error {
// Stop the embedded output
func (n *NotifyOutput) Stop() error {
close(n.eventChan)
close(n.errorChan)
return nil
}

// ErrorChan returns the input error channel
func (n *NotifyOutput) ErrorChan() chan error {
return n.errorChan
// SetErrorChan sets the error channel
func (n *NotifyOutput) SetErrorChan(ch chan<- error) {
n.errorChan = ch
}

// InputChan returns the input event channel
Expand Down
10 changes: 4 additions & 6 deletions output/push/push.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ import (
)

type PushOutput struct {
errorChan chan error
errorChan chan<- error
eventChan chan event.Event
logger plugin.Logger
accessToken string
Expand All @@ -54,7 +54,6 @@ type PushPayload struct {

func New(options ...PushOptionFunc) *PushOutput {
p := &PushOutput{
errorChan: make(chan error),
eventChan: make(chan event.Event, 10),
}
for _, option := range options {
Expand Down Expand Up @@ -286,13 +285,12 @@ func (p *PushOutput) GetProjectId() error {
// Stop the embedded output
func (p *PushOutput) Stop() error {
close(p.eventChan)
close(p.errorChan)
return nil
}

// ErrorChan returns the input error channel
func (p *PushOutput) ErrorChan() chan error {
return p.errorChan
// SetErrorChan sets the error channel
func (p *PushOutput) SetErrorChan(ch chan<- error) {
p.errorChan = ch
}

// InputChan returns the input event channel
Expand Down
10 changes: 4 additions & 6 deletions output/webhook/webhook.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ const (
)

type WebhookOutput struct {
errorChan chan error
errorChan chan<- error
eventChan chan event.Event
logger plugin.Logger
format string
Expand All @@ -51,7 +51,6 @@ type WebhookOutput struct {

func New(options ...WebhookOptionFunc) *WebhookOutput {
w := &WebhookOutput{
errorChan: make(chan error),
eventChan: make(chan event.Event, 10),
format: "adder",
url: "http://localhost:3000",
Expand Down Expand Up @@ -303,13 +302,12 @@ func (w *WebhookOutput) SendWebhook(e *event.Event) error {
// Stop the embedded output
func (w *WebhookOutput) Stop() error {
close(w.eventChan)
close(w.errorChan)
return nil
}

// ErrorChan returns the input error channel
func (w *WebhookOutput) ErrorChan() chan error {
return w.errorChan
// SetErrorChan sets the error channel
func (w *WebhookOutput) SetErrorChan(ch chan<- error) {
w.errorChan = ch
}

// InputChan returns the input event channel
Expand Down
Loading