OCPP: coalesce BootNotification channel to fix reboot-loop hang (#30137)

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

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) {

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() {