diff --git a/charger/ocpp/cp_core.go b/charger/ocpp/cp_core.go index 8ff4f7236..d7f544100 100644 --- a/charger/ocpp/cp_core.go +++ b/charger/ocpp/cp_core.go @@ -28,11 +28,19 @@ func (cp *CP) OnBootNotification(request *core.BootNotificationRequest) (*core.B // mark charge point as ready for communication cp.connect(true) - // notify channel for Setup (initial) or monitorReboot (reconnection) + // Notify the reboot monitor (and the initial Setup). The channel is + // buffered (size 1) and coalescing: if an older notification is still + // queued, drop it so the consumer always re-initializes against the most + // recent BootNotification. This matters when a charge point reboot-loops + // (e.g. issue #30113) faster than the monitor consumes notifications, or + // before the monitor has started at all - the channel must never overflow. + select { + case <-cp.bootNotificationRequestC: + default: + } select { case cp.bootNotificationRequestC <- request: default: - cp.log.DEBUG.Printf("boot notification channel full, discarding") } return res, nil diff --git a/charger/ocpp/cp_core_test.go b/charger/ocpp/cp_core_test.go index 59553ae81..b68624ee8 100644 --- a/charger/ocpp/cp_core_test.go +++ b/charger/ocpp/cp_core_test.go @@ -1,6 +1,7 @@ package ocpp import ( + "fmt" "sync/atomic" "testing" "time" @@ -123,18 +124,19 @@ func TestDisconnectCancelsTimer(t *testing.T) { "should not be connected after cancelled timer") } -func TestBootNotificationChannelFull(t *testing.T) { +func TestBootNotificationChannelCoalesces(t *testing.T) { log := util.NewLogger("test") cp := NewChargePoint(log, "test-cp") - // fill the channel (buffer size 1) + // pre-fill the channel (buffer size 1) with a stale notification cp.bootNotificationRequestC <- &core.BootNotificationRequest{ - ChargePointModel: "First", + ChargePointModel: "Stale", } - // second notification should be dropped (channel full, non-blocking send) + // a fresh BootNotification must replace the stale one rather than be + // discarded - the consumer needs the charge point's current state bootReq := &core.BootNotificationRequest{ - ChargePointModel: "Second", + ChargePointModel: "Fresh", ChargePointVendor: "TestVendor", } @@ -142,13 +144,46 @@ func TestBootNotificationChannelFull(t *testing.T) { require.NoError(t, err) assert.Equal(t, core.RegistrationStatusAccepted, res.Status) - // result should still be updated even though channel was full assert.Equal(t, bootReq, cp.BootNotificationResult) assert.True(t, cp.Connected()) - // channel should have the first message (second was dropped) + // channel must hold the freshest notification, exactly once req := <-cp.bootNotificationRequestC - assert.Equal(t, "First", req.ChargePointModel) + assert.Equal(t, "Fresh", req.ChargePointModel) + + select { + case extra := <-cp.bootNotificationRequestC: + t.Fatalf("channel should hold exactly one notification, got extra: %s", extra.ChargePointModel) + default: + } +} + +// TestBootNotificationRebootLoopKeepsLatest reproduces the EN+ reboot loop from +// issue #30113: a charge point reconnects repeatedly, sending a BootNotification +// each time, before the reboot monitor consumes any of them. The buffered +// channel must never overflow and must always retain the most recent +// notification so a later Setup re-runs against the charge point's real state. +func TestBootNotificationRebootLoopKeepsLatest(t *testing.T) { + log := util.NewLogger("test") + cp := NewChargePoint(log, "test-cp") + + for i := range 5 { + _, err := cp.OnBootNotification(&core.BootNotificationRequest{ + ChargePointModel: "EN+", + FirmwareVersion: fmt.Sprintf("1.1.%d", i), + }) + require.NoError(t, err) + } + + // exactly one notification queued, and it is the latest + req := <-cp.bootNotificationRequestC + assert.Equal(t, "1.1.4", req.FirmwareVersion) + + select { + case extra := <-cp.bootNotificationRequestC: + t.Fatalf("channel must buffer at most one notification, got extra: %s", extra.FirmwareVersion) + default: + } } func TestReconnectAfterReboot(t *testing.T) { diff --git a/charger/ocpp_test.go b/charger/ocpp_test.go index c68df8e73..dc4e14f47 100644 --- a/charger/ocpp_test.go +++ b/charger/ocpp_test.go @@ -58,9 +58,12 @@ func (suite *ocppTestSuite) TearDownSuite() { } func (suite *ocppTestSuite) startChargePoint(id string, connectorId int) (ocpp16.ChargePoint, *ocppj.Client) { - // Buffered generously: handlers in ocpp_test_handler.go dispatch sends via - // `go func() { triggerC <- … }()`, so a small buffer leaks one extra - // goroutine per concurrent trigger until the drain catches up. + // Buffered generously: the handlers in ocpp_test_handler.go send to + // triggerC synchronously (via defer) on the charge point's WebSocket + // read-loop goroutine. If that send blocks, the read loop cannot deliver + // the CALL_RESULT for the CP→CS request the drain goroutine is waiting + // on, deadlocking the test. The buffer keeps the send from blocking + // until the drain catches up. handler := &ChargePointHandler{ triggerC: make(chan remotetrigger.MessageTrigger, 16), } @@ -79,8 +82,9 @@ func (suite *ocppTestSuite) startChargePoint(id string, connectorId int) (ocpp16 // let cs handle the trigger messages; exit on done so we do not leak a // drain goroutine into subsequent subtests on the shared ocpp.Instance(). - // Cannot close triggerC because the async senders in ocpp_test_handler.go - // would panic on `send on closed channel`. + // triggerC is deliberately left open: the handlers in ocpp_test_handler.go + // send to it from the charge point's read loop, which we do not synchronize + // with on shutdown, so closing it could panic on `send on closed channel`. done := make(chan struct{}) finished := make(chan struct{}) go func() {