fix(ocpp): take up a running 1.6 transaction after a Core restart

.changeset/ocpp-resume-transaction.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+An OCPP 1.6 charger that keeps charging while Core restarts now has its session energy counted again. Before, Core forgot the running transaction, so session energy stayed at 0 until the car was unplugged, and a stop then reported the charger's lifetime meter total as the session. Core now takes up the transaction from the first meter value the charger sends for it and counts energy from that reading. Energy from before the restart is still lost. Transaction ids now start from the clock, so a restart no longer hands out an id that a charger may still hold.

go/internal/ocpp/handlers.go

@@ -98,6 +98,9 @@ type chargerState struct {
 	// transactionRef is the 2.0.1 transaction id, which is a string rather
 	// than the int transactionID above. Empty on the 1.6 path.
 	transactionRef string
+	// stoppedTxID is the last transaction this charger stopped. A late meter
+	// value for it must not take it up again.
+	stoppedTxID int
 	// vendor and model come from BootNotification and exist so the UI can
 	// label a charger with what it actually is rather than its URL segment.
 	// serial is the hardware-stable half of its identity — what the device
@@ -146,10 +149,22 @@ func NewHandler(tel *telemetry.Store, heartbeatIntervalS int) *Handler {
 		approved:            map[string]bool{},
 		probing:             map[string]bool{},
 		identityProbeTiming: defaultIdentityProbeTiming,
-		nextTxID:            1,
+		nextTxID:            firstTxID(time.Now()),
 	}
 }
 
+// txIDEpoch keeps clock-based transaction ids far below the 32-bit limit
+// many chargers store them in.
+var txIDEpoch = time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+
+// firstTxID starts a run's transaction ids at the seconds since txIDEpoch.
+// A charger keeps its transaction id while Core restarts, so a restart must
+// not hand that id to another transaction. The start moves on one per
+// second, and a home starts far fewer transactions than that.
+func firstTxID(now time.Time) int {
+	return max(1, int(now.Sub(txIDEpoch)/time.Second))
+}
+
 // SetApprovedIDs declares which charger ids are part of the site — the ids
 // loadpoints name in config. Everything else that connects is quarantined as
 // pending: accepted at the protocol level so the operator can see it in the
@@ -612,6 +627,9 @@ func (h *Handler) OnMeterValues(id string, req *core.MeterValuesRequest) (*core.
 				if sv.Unit == types.UnitOfMeasureKWh {
 					val *= 1000
 				}
+				if s.transactionID < 0 && sv.Phase == "" && req.TransactionId != nil {
+					h.takeUpTransactionLocked(id, s, *req.TransactionId, val)
+				}
 				if s.transactionID >= 0 {
 					s.sessionMeterWh = val - s.sessionStartMeterWh
 				}
@@ -628,6 +646,21 @@ func (h *Handler) OnMeterValues(id string, req *core.MeterValuesRequest) (*core.
 	return core.NewMeterValuesConfirmation(), nil
 }
 
+// takeUpTransactionLocked adopts a transaction that started before this run:
+// the charger kept it, and its meter values name it, while Core restarted.
+// The start meter reading went with the old process, so session energy
+// counts from this first reading. The caller holds h.mu.
+func (h *Handler) takeUpTransactionLocked(id string, s *chargerState, txID int, meterWh float64) {
+	if txID <= 0 || txID == s.stoppedTxID {
+		return
+	}
+	s.transactionID = txID
+	s.sessionStartMeterWh = meterWh
+	s.sessionMeterWh = 0
+	h.nextTxID = max(h.nextTxID, txID+1)
+	slog.Info("OCPP running transaction taken up", "charger", id, "txid", txID, "meter_wh", meterWh)
+}
+
 func (h *Handler) OnStartTransaction(id string, req *core.StartTransactionRequest) (*core.StartTransactionConfirmation, error) {
 	h.mu.Lock()
 	txID := h.nextTxID
@@ -658,7 +691,14 @@ func (h *Handler) OnStartTransaction(id string, req *core.StartTransactionReques
 func (h *Handler) OnStopTransaction(id string, req *core.StopTransactionRequest) (*core.StopTransactionConfirmation, error) {
 	s := h.state(id)
 	h.mu.Lock()
-	sessionWh := float64(req.MeterStop) - s.sessionStartMeterWh
+	// With no start and no meter value seen for the transaction, MeterStop is
+	// only the meter's lifetime total, not the session's energy.
+	known := s.transactionID >= 0
+	sessionWh := s.sessionMeterWh
+	if known {
+		sessionWh = float64(req.MeterStop) - s.sessionStartMeterWh
+	}
+	s.stoppedTxID = req.TransactionId
 	s.transactionID = -1
 	s.charging = false
 	// A charger may report Available before the stop. The car is gone then.
@@ -672,7 +712,9 @@ func (h *Handler) OnStopTransaction(id string, req *core.StopTransactionRequest)
 		"charger", id, "txid", req.TransactionId,
 		"session_wh", sessionWh, "reason", req.Reason)
 	h.pushReading(id, s)
-	h.telMetric(id, "ev_session_wh", sessionWh, "Wh")
+	if known {
+		h.telMetric(id, "ev_session_wh", sessionWh, "Wh")
+	}
 	h.telSuccess(id)
 	return core.NewStopTransactionConfirmation(), nil
 }

go/internal/ocpp/transaction_resume_test.go

@@ -0,0 +1,143 @@
+package ocpp
+
+import (
+	"encoding/json"
+	"math"
+	"testing"
+	"time"
+
+	"github.com/lorenzodonini/ocpp-go/ocpp1.6/core"
+	"github.com/lorenzodonini/ocpp-go/ocpp1.6/types"
+
+	"github.com/srcfl/ftw/go/internal/telemetry"
+)
+
+// restartedHandler is Core just after a restart: a fresh handler and an
+// adopted charger that reconnected while its car kept charging.
+func restartedHandler(t *testing.T, tel *telemetry.Store, ids ...string) *Handler {
+	t.Helper()
+	h := NewHandler(tel, 60)
+	h.SetApprovedIDs(ids)
+	for _, id := range ids {
+		h.OnConnect(id)
+		h.OnStatusNotification(id, &core.StatusNotificationRequest{ConnectorId: 1, Status: core.ChargePointStatusCharging})
+	}
+	return h
+}
+
+func meterRegister(h *Handler, id string, tx int, wh string) {
+	h.OnMeterValues(id, &core.MeterValuesRequest{ConnectorId: 1, TransactionId: &tx, MeterValue: []types.MeterValue{{
+		Timestamp:    types.NewDateTime(time.Now()),
+		SampledValue: []types.SampledValue{{Value: wh, Measurand: types.MeasurandEnergyActiveImportRegister, Unit: types.UnitOfMeasureWh}},
+	}}})
+}
+
+func readingSessionWh(t *testing.T, tel *telemetry.Store, id string) float64 {
+	t.Helper()
+	r := tel.Get(id, telemetry.DerEV)
+	if r == nil {
+		t.Fatalf("no EV reading for %s", id)
+	}
+	var d struct {
+		SessionWh float64 `json:"session_wh"`
+	}
+	if err := json.Unmarshal(r.Data, &d); err != nil {
+		t.Fatal(err)
+	}
+	return d.SessionWh
+}
+
+// The charger keeps its transaction while Core restarts and goes on sending
+// meter values for it. Core takes the transaction up from the first of them,
+// so session energy counts again instead of staying at 0 until unplug.
+func TestOCPP16PicksUpRunningTransactionAfterRestart(t *testing.T) {
+	tel := telemetry.NewStore()
+	h := restartedHandler(t, tel, "charger")
+
+	meterRegister(h, "charger", 41, "15200")
+	meterRegister(h, "charger", 41, "15700")
+
+	if v := h.Snapshot()["charger"]; v.TxID != 41 || v.SessionWh != 500 {
+		t.Fatalf("after restart: tx %d, session %.0f Wh; want tx 41, 500 Wh", v.TxID, v.SessionWh)
+	}
+	if got := readingSessionWh(t, tel, "charger"); got != 500 {
+		t.Fatalf("reading session_wh = %.0f, want 500", got)
+	}
+
+	h.OnStopTransaction("charger", &core.StopTransactionRequest{TransactionId: 41, MeterStop: 15900})
+	if v := h.Snapshot()["charger"]; v.TxID != -1 || v.SessionWh != 700 {
+		t.Fatalf("after stop: tx %d, session %.0f Wh; want tx -1, 700 Wh", v.TxID, v.SessionWh)
+	}
+}
+
+// A stop for a transaction Core never saw start carries only the meter's
+// lifetime total. That is not the session's energy.
+func TestOCPP16StopOfUnseenTransactionKeepsLifetimeMeterOut(t *testing.T) {
+	tel := telemetry.NewStore()
+	h := restartedHandler(t, tel, "charger")
+
+	h.OnStopTransaction("charger", &core.StopTransactionRequest{TransactionId: 41, MeterStop: 1234567})
+
+	if v := h.Snapshot()["charger"]; v.SessionWh != 0 {
+		t.Fatalf("session_wh = %.0f, want 0: MeterStop is the lifetime register", v.SessionWh)
+	}
+}
+
+// A late meter value for a transaction that has stopped must not start it
+// again: session energy would fall to 0 while the car is still plugged in.
+func TestOCPP16StoppedTransactionIsNotTakenUpAgain(t *testing.T) {
+	tel := telemetry.NewStore()
+	h := restartedHandler(t, tel, "charger")
+	conf, err := h.OnStartTransaction("charger", &core.StartTransactionRequest{ConnectorId: 1, IdTag: "tag", MeterStart: 1000})
+	if err != nil {
+		t.Fatal(err)
+	}
+	h.OnStopTransaction("charger", &core.StopTransactionRequest{TransactionId: conf.TransactionId, MeterStop: 8500})
+
+	meterRegister(h, "charger", conf.TransactionId, "8500")
+
+	if v := h.Snapshot()["charger"]; v.TxID != -1 || v.SessionWh != 7500 {
+		t.Fatalf("tx %d, session %.0f Wh; want tx -1, 7500 Wh", v.TxID, v.SessionWh)
+	}
+}
+
+// Every transaction id goes to one transaction, across restarts too.
+func TestOCPP16TransactionIDsDoNotRepeatAfterRestart(t *testing.T) {
+	tel := telemetry.NewStore()
+	before := restartedHandler(t, tel, "a", "b")
+	old, err := before.OnStartTransaction("a", &core.StartTransactionRequest{ConnectorId: 1, IdTag: "tag", MeterStart: 100})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if old.TransactionId == 1 {
+		t.Fatalf("first id after a start is 1, which every earlier run also handed out")
+	}
+
+	// Core restarts within the same second. Charger a reports its running
+	// transaction; then b starts one.
+	after := restartedHandler(t, tel, "a", "b")
+	meterRegister(after, "a", old.TransactionId, "400")
+	next, err := after.OnStartTransaction("b", &core.StartTransactionRequest{ConnectorId: 1, IdTag: "tag", MeterStart: 0})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if next.TransactionId == old.TransactionId {
+		t.Fatalf("b got transaction id %d, which a still holds", next.TransactionId)
+	}
+}
+
+// A run starts its ids at the seconds since txIDEpoch, so a later run starts
+// above the ids an earlier one used unless it started more transactions than
+// it ran seconds.
+func TestOCPP16FirstTransactionID(t *testing.T) {
+	t0 := time.Date(2026, 10, 21, 13, 1, 49, 0, time.UTC)
+	if firstTxID(t0.Add(20*time.Second))-firstTxID(t0) != 20 {
+		t.Fatalf("ids do not move on one per second")
+	}
+	if got := firstTxID(time.Unix(0, 0)); got != 1 {
+		t.Fatalf("clock before the epoch: first id %d, want 1", got)
+	}
+	if got := firstTxID(time.Date(2090, 1, 1, 0, 0, 0, 0, time.UTC)); got >= math.MaxInt32 {
+		t.Fatalf("first id %d in 2090 does not fit the 32 bits chargers store it in", got)
+	}
+}