diff --git a/charger/ocpp/forwarder.go b/charger/ocpp/forwarder.go index 16820671c..89365391a 100644 --- a/charger/ocpp/forwarder.go +++ b/charger/ocpp/forwarder.go @@ -15,6 +15,7 @@ import ( "crypto/x509" "encoding/base64" "encoding/json" + "errors" "fmt" "net/http" "slices" @@ -23,6 +24,7 @@ import ( "sync" "time" + "github.com/cenkalti/backoff/v4" "github.com/coder/websocket" "github.com/evcc-io/evcc/util" "github.com/lorenzodonini/ocpp-go/ocpp1.6/core" @@ -102,7 +104,7 @@ func ApplyForwarderRules(rules []ForwarderRule) { pendingMu.Lock() pendingMsgs[r.StationID] = nil pendingMu.Unlock() - go dialUpstreamSidecar(r.StationID, r) + go runUpstreamSidecar(r.StationID, r) } else { go validateUpstream(r.StationID, r) } @@ -111,19 +113,13 @@ func ApplyForwarderRules(rules []ForwarderRule) { notifyUpdated() } -// validateUpstream test-dials a rule's upstream and records/clears the error so -// the UI reflects unreachable hosts. -func validateUpstream(id string, rule ForwarderRule) { - upstreamBase := strings.TrimRight(rule.UpstreamURL, "/") - upstreamPath := rule.upstreamPath(id) - +// dialUpstream opens a websocket connection to the rule's upstream server for the given charger. +func dialUpstream(id string, rule ForwarderRule) (*websocket.Conn, error) { tlsConfig := &tls.Config{InsecureSkipVerify: rule.Insecure} if rule.CaCert != "" { caCertPool := x509.NewCertPool() if ok := caCertPool.AppendCertsFromPEM([]byte(rule.CaCert)); !ok { - recordForwarderError(id, "invalid CA certificate") - notifyUpdated() - return + return nil, errors.New("invalid CA certificate") } tlsConfig.RootCAs = caCertPool } @@ -134,12 +130,20 @@ func validateUpstream(id string, rule ForwarderRule) { } ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - conn, _, err := websocket.Dial(ctx, upstreamBase+upstreamPath, &websocket.DialOptions{ + defer cancel() + + conn, _, err := websocket.Dial(ctx, strings.TrimRight(rule.UpstreamURL, "/")+rule.upstreamPath(id), &websocket.DialOptions{ Subprotocols: []string{"ocpp1.6"}, HTTPHeader: header, HTTPClient: &http.Client{Transport: &http.Transport{TLSClientConfig: tlsConfig}}, }) - cancel() + return conn, err +} + +// validateUpstream test-dials a rule's upstream and records/clears the error so +// the UI reflects unreachable hosts. +func validateUpstream(id string, rule ForwarderRule) { + conn, err := dialUpstream(id, rule) // a charger may have connected meanwhile; its sidecar is authoritative sidecarsMu.Lock() @@ -190,10 +194,9 @@ func init() { // sidecar holds an upstream connection for a single charger. type sidecar struct { - chargerID string - upstreamURL string - rule ForwarderRule // rule used to dial; detects param changes - conn *websocket.Conn + chargerID string + rule ForwarderRule // rule used to dial; detects param changes + conn *websocket.Conn // message IDs of upstream-initiated Calls; the charger's reply is routed back to upstream pendingUpstreamCallsMu sync.Mutex @@ -223,8 +226,58 @@ var ( // on connect so BootNotification reaches upstream pendingMu sync.Mutex pendingMsgs = make(map[string][][]byte) + + // last BootNotification per connected charger, replayed to upstream when a + // sidecar (re)connects mid-session so upstream sees a boot before transactions + bootMu sync.Mutex + lastBoot = make(map[string][]byte) ) +// reconnect backoff bounds; variables for testing +var ( + reconnectInitialDelay = 5 * time.Second + reconnectMaxDelay = 5 * time.Minute +) + +// runUpstreamSidecar dials a charger's upstream sidecar and re-dials with +// exponential backoff for as long as the charger stays connected and the rule +// remains in effect. +func runUpstreamSidecar(id string, rule ForwarderRule) { + bo := backoff.NewExponentialBackOff( + backoff.WithInitialInterval(reconnectInitialDelay), + backoff.WithMaxInterval(reconnectMaxDelay), + backoff.WithMaxElapsedTime(0), + ) + + for { + if !dialUpstreamSidecar(id, rule) { + delay := bo.NextBackOff() + forwarderLog.DEBUG.Printf("forwarder: reconnecting upstream for %s in %v", id, delay) + time.Sleep(delay) + continue + } + + bo.Reset() + + if current, ok := resolveRule(id); !ok || current.UpstreamURL == "" || !current.sameConnection(rule) { + return + } + + sidecarsMu.Lock() + connected := connectedChargers[id] + _, active := sidecars[id] + sidecarsMu.Unlock() + + if !connected || active { + return + } + + pendingMu.Lock() + pendingMsgs[id] = nil + pendingMu.Unlock() + } +} + // resolveRule returns the forwarding rule for chargerID, or the "*" fallback. func resolveRule(chargerID string) (ForwarderRule, bool) { forwarderMu.RLock() @@ -261,50 +314,22 @@ func onChargerConnect(ch ws.Channel) { pendingMsgs[id] = nil pendingMu.Unlock() - go dialUpstreamSidecar(id, rule) + go runUpstreamSidecar(id, rule) } -func dialUpstreamSidecar(id string, rule ForwarderRule) { - upstreamBase := strings.TrimRight(rule.UpstreamURL, "/") - upstreamPath := rule.upstreamPath(id) - - var header http.Header - if rule.Username != "" || rule.Password != "" { - header = authHeader(rule.Username, rule.Password) - } - - tlsConfig := &tls.Config{InsecureSkipVerify: rule.Insecure} - if rule.CaCert != "" { - caCertPool := x509.NewCertPool() - if ok := caCertPool.AppendCertsFromPEM([]byte(rule.CaCert)); !ok { - forwarderLog.WARN.Printf("forwarder: failed to parse CA cert for %s; forwarding disabled", id) - recordForwarderError(id, "invalid CA certificate") - notifyUpdated() - drainPendingWithErrors(id, nil) - return - } - tlsConfig.RootCAs = caCertPool - } - - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - conn, _, err := websocket.Dial(ctx, upstreamBase+upstreamPath, &websocket.DialOptions{ - Subprotocols: []string{"ocpp1.6"}, - HTTPHeader: header, - HTTPClient: &http.Client{Transport: &http.Transport{TLSClientConfig: tlsConfig}}, - }) - cancel() +func dialUpstreamSidecar(id string, rule ForwarderRule) bool { + conn, err := dialUpstream(id, rule) if err != nil { - forwarderLog.WARN.Printf("forwarder: dial upstream for %s: %v; forwarding disabled", id, err) + forwarderLog.WARN.Printf("forwarder: dial upstream for %s: %v", id, err) recordForwarderError(id, err.Error()) notifyUpdated() drainPendingWithErrors(id, nil) - return + return false } conn.SetReadLimit(-1) // no limit; OCPP frames can be large sc := &sidecar{ chargerID: id, - upstreamURL: upstreamBase, rule: rule, conn: conn, pendingUpstreamCalls: make(map[string]struct{}), @@ -314,12 +339,35 @@ func dialUpstreamSidecar(id string, rule ForwarderRule) { // install sidecar and drain the pending buffer sidecarsMu.Lock() pendingMu.Lock() + old := sidecars[id] sidecars[id] = sc buffered := pendingMsgs[id] delete(pendingMsgs, id) pendingMu.Unlock() sidecarsMu.Unlock() + // a concurrent dial may have installed a sidecar meanwhile; this one wins + if old != nil { + old.conn.CloseNow() + } + + // replay the charger's BootNotification (with a fresh message id) when the + // buffer doesn't already carry one, i.e. the sidecar (re)connects mid-session; + // upstream's reply is discarded as evcc has long answered the charger + if !slices.ContainsFunc(buffered, func(frame []byte) bool { + msgType, _, action, err := parseOCPPFrame(frame) + return err == nil && msgType == ocppj.CALL && action == "BootNotification" + }) { + bootMu.Lock() + boot := lastBoot[id] + bootMu.Unlock() + if boot != nil { + if frame, err := withMessageID(boot, fmt.Sprintf("evcc-boot-%d", time.Now().UnixNano())); err == nil { + buffered = append([][]byte{frame}, buffered...) + } + } + } + // flush buffered Calls; register relay actions so upstream's reply routes back. // CallResults/Errors are skipped: evcc already answered them. flushed := 0 @@ -333,12 +381,20 @@ func dialUpstreamSidecar(id string, rule ForwarderRule) { sc.pendingChargerCalls[msgID] = struct{}{} sc.pendingChargerCallsMu.Unlock() } - if err := conn.Write(context.Background(), websocket.MessageText, frame); err != nil { - forwarderLog.ERROR.Printf("forwarder: write buffered frame to upstream for %s: %v", id, err) + writeCtx, writeCancel := context.WithTimeout(context.Background(), 10*time.Second) + writeErr := conn.Write(writeCtx, websocket.MessageText, frame) + writeCancel() + if writeErr != nil { + forwarderLog.ERROR.Printf("forwarder: write buffered frame to upstream for %s: %v", id, writeErr) conn.CloseNow() - recordForwarderError(id, err.Error()) + sidecarsMu.Lock() + if sidecars[id] == sc { + delete(sidecars, id) + } + sidecarsMu.Unlock() + recordForwarderError(id, writeErr.Error()) notifyUpdated() - return + return false } flushed++ } @@ -349,9 +405,11 @@ func dialUpstreamSidecar(id string, rule ForwarderRule) { clearForwarderError(id) notifyUpdated() - forwarderLog.INFO.Printf("forwarder: %s → %s", id, upstreamBase+upstreamPath) + forwarderLog.INFO.Printf("forwarder: %s → %s", id, strings.TrimRight(rule.UpstreamURL, "/")+rule.upstreamPath(id)) - sc.readFromUpstream(rule.ReadOnly) + sc.readFromUpstream() + + return true } // drainPendingWithErrors discards charger id's pending buffer, sending a CallError @@ -403,10 +461,13 @@ func drainPendingWithErrors(id string, sc *sidecar) { func onChargerDisconnect(ch ws.Channel) { id := ch.ID() - // discard pending buffer + // discard pending buffer and cached boot frame pendingMu.Lock() delete(pendingMsgs, id) pendingMu.Unlock() + bootMu.Lock() + delete(lastBoot, id) + bootMu.Unlock() sidecarsMu.Lock() delete(connectedChargers, id) @@ -439,6 +500,13 @@ func onChargerMessage(ch ws.Channel, data []byte) bool { return false } + // remember the charger's boot frame for replay on sidecar (re)connect + if msgType == ocppj.CALL && action == "BootNotification" { + bootMu.Lock() + lastBoot[id] = slices.Clone(data) + bootMu.Unlock() + } + sidecarsMu.Lock() sc := sidecars[id] sidecarsMu.Unlock() @@ -506,7 +574,7 @@ func onChargerMessage(ch ws.Channel, data []byte) bool { // readFromUpstream relays frames from upstream: Calls are injected into the // charger (reply routed back), responses to bypassed charger Calls are relayed // to the charger, others discarded. Calls are rejected in read-only mode. -func (sc *sidecar) readFromUpstream(readOnly bool) { +func (sc *sidecar) readFromUpstream() { defer func() { sidecarsMu.Lock() if sidecars[sc.chargerID] == sc { @@ -541,7 +609,8 @@ func (sc *sidecar) readFromUpstream(readOnly bool) { switch msgType { case ocppj.CALL: - if readOnly { + // resolve the current rule so a runtime ReadOnly toggle applies live + if rule, ok := resolveRule(sc.chargerID); ok && rule.ReadOnly { forwarderLog.DEBUG.Printf("forwarder: blocking upstream call %s in read-only session %s", msgID, sc.chargerID) errFrame, _ := (&ocppj.CallError{ MessageTypeId: ocppj.CALL_ERROR, @@ -724,6 +793,23 @@ func extractMeterValueSampleInterval(msg []byte) (time.Duration, bool) { return time.Duration(secs) * time.Second, true } +// withMessageID returns a copy of a raw OCPP frame with its message id replaced. +func withMessageID(frame []byte, msgID string) ([]byte, error) { + var parts []json.RawMessage + if err := json.Unmarshal(frame, &parts); err != nil { + return nil, err + } + if len(parts) < 2 { + return nil, fmt.Errorf("invalid OCPP frame") + } + id, err := json.Marshal(msgID) + if err != nil { + return nil, err + } + parts[1] = id + return json.Marshal(parts) +} + // parseOCPPFrame extracts the message type, id and (for Calls) action from a raw frame. func parseOCPPFrame(msg []byte) (msgType ocppj.MessageType, msgID string, action string, err error) { var frame []json.RawMessage diff --git a/charger/ocpp/forwarder_test.go b/charger/ocpp/forwarder_test.go new file mode 100644 index 000000000..7eded2504 --- /dev/null +++ b/charger/ocpp/forwarder_test.go @@ -0,0 +1,174 @@ +package ocpp + +import ( + "crypto/tls" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/coder/websocket" + "github.com/lorenzodonini/ocpp-go/ocppj" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const testBootFrame = `[2,"orig-id","BootNotification",{"chargePointVendor":"v","chargePointModel":"m"}]` + +func TestForwarderWithMessageID(t *testing.T) { + frame, err := withMessageID([]byte(testBootFrame), "new-id") + require.NoError(t, err) + + msgType, msgID, action, err := parseOCPPFrame(frame) + require.NoError(t, err) + assert.Equal(t, ocppj.CALL, msgType) + assert.Equal(t, "new-id", msgID) + assert.Equal(t, "BootNotification", action) + assert.Contains(t, string(frame), `"chargePointVendor":"v"`) + + _, err = withMessageID([]byte(`{}`), "new-id") + assert.Error(t, err) +} + +// fakeChannel implements ws.Channel for hook tests +type fakeChannel string + +func (c fakeChannel) ID() string { return string(c) } +func (c fakeChannel) RemoteAddr() net.Addr { return nil } +func (c fakeChannel) TLSConnectionState() *tls.ConnectionState { return nil } +func (c fakeChannel) IsConnected() bool { return true } + +func TestForwarderBootNotificationCache(t *testing.T) { + const id = "fwd-boot-cache-test" + + onChargerMessage(fakeChannel(id), []byte(testBootFrame)) + + bootMu.Lock() + boot := lastBoot[id] + bootMu.Unlock() + assert.Equal(t, testBootFrame, string(boot)) + + onChargerDisconnect(fakeChannel(id)) + + bootMu.Lock() + boot = lastBoot[id] + bootMu.Unlock() + assert.Nil(t, boot) +} + +// forwarderTestSetup marks a charger as connected with a cached boot frame and +// fast reconnect, returning a cleanup function. +func forwarderTestSetup(t *testing.T, id string) func() { + t.Helper() + + delay := reconnectInitialDelay + reconnectInitialDelay = 10 * time.Millisecond + + sidecarsMu.Lock() + connectedChargers[id] = true + sidecarsMu.Unlock() + + bootMu.Lock() + lastBoot[id] = []byte(testBootFrame) + bootMu.Unlock() + + return func() { + ApplyForwarderRules(nil) + sidecarsMu.Lock() + delete(connectedChargers, id) + sidecarsMu.Unlock() + bootMu.Lock() + delete(lastBoot, id) + bootMu.Unlock() + reconnectInitialDelay = delay + } +} + +// expectBootReplay asserts that a replayed BootNotification with a fresh message id arrives. +func expectBootReplay(t *testing.T, frames <-chan []byte) { + t.Helper() + + select { + case data := <-frames: + msgType, msgID, action, err := parseOCPPFrame(data) + require.NoError(t, err) + assert.Equal(t, ocppj.CALL, msgType) + assert.Equal(t, "BootNotification", action) + assert.NotEqual(t, "orig-id", msgID) + case <-time.After(5 * time.Second): + t.Fatal("timeout waiting for upstream reconnect") + } +} + +func TestForwarderUpstreamReconnect(t *testing.T) { + const id = "fwd-reconnect-test" + + var conns atomic.Int32 + frames := make(chan []byte, 1) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + c, err := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{"ocpp1.6"}}) + if err != nil { + return + } + if conns.Add(1) == 1 { + // simulate upstream dropping the connection without close handshake + c.CloseNow() + return + } + if _, data, err := c.Read(r.Context()); err == nil { + select { + case frames <- data: + default: + } + } + c.CloseNow() + })) + defer srv.Close() + + cleanup := forwarderTestSetup(t, id) + defer cleanup() + + ApplyForwarderRules([]ForwarderRule{{StationID: id, UpstreamURL: "ws" + strings.TrimPrefix(srv.URL, "http")}}) + + expectBootReplay(t, frames) + assert.GreaterOrEqual(t, conns.Load(), int32(2)) +} + +func TestForwarderDialRetry(t *testing.T) { + const id = "fwd-dial-retry-test" + + var reqs atomic.Int32 + frames := make(chan []byte, 1) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if reqs.Add(1) <= 2 { + // reject the websocket upgrade to simulate an unreachable upstream + http.Error(w, "unavailable", http.StatusServiceUnavailable) + return + } + c, err := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{"ocpp1.6"}}) + if err != nil { + return + } + if _, data, err := c.Read(r.Context()); err == nil { + select { + case frames <- data: + default: + } + } + c.CloseNow() + })) + defer srv.Close() + + cleanup := forwarderTestSetup(t, id) + defer cleanup() + + ApplyForwarderRules([]ForwarderRule{{StationID: id, UpstreamURL: "ws" + strings.TrimPrefix(srv.URL, "http")}}) + + expectBootReplay(t, frames) + assert.GreaterOrEqual(t, reqs.Load(), int32(3)) +} diff --git a/docs/agents/ocpp-forwarder.md b/docs/agents/ocpp-forwarder.md index 173bf356a..e141589ed 100644 --- a/docs/agents/ocpp-forwarder.md +++ b/docs/agents/ocpp-forwarder.md @@ -41,4 +41,8 @@ When a rule sets `ReadOnly`, upstream may observe but cannot control the charger Frames that arrive from a charger before its sidecar finishes dialling are buffered (`pendingMsgs`) and flushed in order once the sidecar connects, so early messages such as `BootNotification` still reach upstream. If the dial fails or upstream drops mid-session, any buffered or in-flight relay Calls are answered to the charger with a `CallError` so it is not left hanging, and the failure is surfaced to the UI via `forwarderErrors`. +When the upstream connection fails while the charger stays connected, the sidecar is re-dialled automatically with exponential backoff (`runUpstreamSidecar`, 5s up to 5min; a successful session resets the backoff). Between attempts the relay actions fall back to evcc's local handler, so charging continues but upstream misses those transactions. The reconnect loop ends when the charger disconnects, the rule is removed, or its connection parameters change (`ApplyForwarderRules` dials its own sidecar in that case). + +The last `BootNotification` of each connected charger is cached (`lastBoot`). When a sidecar connects mid-session (after an upstream reconnect or a rule added at runtime) and the pending buffer does not already carry a boot frame, the cached frame is replayed to upstream with a fresh message id, since many backends expect a boot before accepting transactions. Upstream's reply to the replay is discarded; evcc answered the charger's original boot long ago. + Rules can be changed at runtime through `ApplyForwarderRules`. Sidecars for removed rules are closed, rules with changed connection parameters are re-dialled, and rules for chargers that are not connected are test-dialled to surface unreachable hosts immediately.