From 954e47f844af26b3722a16c87582c601c2aa0893 Mon Sep 17 00:00:00 2001 From: Tan Le Date: Thu, 30 Jul 2026 12:26:20 +0700 Subject: [PATCH 1/7] feat(rde): port SETTINGS_BYTE request/response bits and message ID MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The firmware implements the full v1.1.x settings byte; the Go port had only bit 0. Adds bit 1 (request), bit 2 (response), the message ID at Buffer[1], and UpdateBitPositionsForHeaderByte to skip it when decoding. Bit 1 is required to parse a §2.2 request payload: it carries POINT_IDs with no DATA_TYPE_ID, so without the flag a decoder cannot know whether to consume 6 data-type bits after each POINT_ID. No-op when both flags are clear, so existing uplinks decode identically. --- codecs/rubixDataEncoding/serialdata.go | 63 ++++++++++++++++++++ codecs/rubixDataEncoding/serialdata_test.go | 65 +++++++++++++++++++++ 2 files changed, 128 insertions(+) create mode 100644 codecs/rubixDataEncoding/serialdata_test.go diff --git a/codecs/rubixDataEncoding/serialdata.go b/codecs/rubixDataEncoding/serialdata.go index 3eeee3b..c6d438d 100644 --- a/codecs/rubixDataEncoding/serialdata.go +++ b/codecs/rubixDataEncoding/serialdata.go @@ -72,3 +72,66 @@ func setPositionalData(serialData *SerialData, set bool) { func hasPositionalData(serialData *SerialData) bool { return serialData.Buffer[0]&1 == 1 } + +// SETTINGS_BYTE bit layout (LORA RAW PROTOCOL §3.3, RDE v1.1.x). Mirrors +// Core/Inc/serialData.hpp in the device firmware; keep the two in step. +// +// bit 0 - positional flag: POINT_IDs are present in the data packets +// bit 1 - request flag: payload is POINT_IDs only, no DATA_TYPE_ID/values +// bit 2 - response flag: payload answers a request +// +// When bit 1 or bit 2 is set, Buffer[1] holds the RDE message ID and the +// first data packet starts one byte later. + +func SetRequestData(serialData *SerialData, set bool) { + serialData.Buffer[0] = BIT_SET(serialData.Buffer[0], set, 1) + SetMessageId(serialData, 0) +} + +func HasRequestData(serialData *SerialData) bool { + return serialData.Buffer[0]&2 == 2 +} + +func SetResponseData(serialData *SerialData, set bool) { + serialData.Buffer[0] = BIT_SET(serialData.Buffer[0], set, 2) + SetMessageId(serialData, 0) +} + +func HasResponseData(serialData *SerialData) bool { + return serialData.Buffer[0]&4 == 4 +} + +func HasPositionalData(serialData *SerialData) bool { + return hasPositionalData(serialData) +} + +// SetMessageId writes the RDE message ID, reserving Buffer[1] if the buffer +// has not grown that far yet. It is a no-op unless a request or response flag +// is set, matching the firmware's setMessageId precondition. +func SetMessageId(serialData *SerialData, id uint8) { + if !HasRequestData(serialData) && !HasResponseData(serialData) { + return + } + if len(serialData.Buffer) > 1 { + serialData.Buffer[1] = id + return + } + serialData.Buffer = append(serialData.Buffer, id) +} + +func GetMessageId(serialData *SerialData) uint8 { + if len(serialData.Buffer) < 2 { + return 0 + } + return serialData.Buffer[1] +} + +// UpdateBitPositionsForHeaderByte advances the read position past the RDE +// message ID byte. Must be called before canDecode/decodeData on any buffer +// that may carry a request or response flag. No-op when both flags are clear, +// so plain uplink decoding is unaffected. +func UpdateBitPositionsForHeaderByte(serialData *SerialData) { + if HasRequestData(serialData) || HasResponseData(serialData) { + serialData.ReadBitPos += 8 + } +} diff --git a/codecs/rubixDataEncoding/serialdata_test.go b/codecs/rubixDataEncoding/serialdata_test.go new file mode 100644 index 0000000..080ccba --- /dev/null +++ b/codecs/rubixDataEncoding/serialdata_test.go @@ -0,0 +1,65 @@ +package rubixDataEncoding + +import "testing" + +func TestRequestFlagRoundTrip(t *testing.T) { + sd := NewSerialData() + if HasRequestData(sd) { + t.Fatal("request flag should start clear") + } + SetRequestData(sd, true) + if !HasRequestData(sd) { + t.Fatal("request flag should be set") + } + if HasResponseData(sd) { + t.Fatal("setting request must not set response") + } + if HasPositionalData(sd) { + t.Fatal("setting request must not set positional") + } +} + +// SetRequestData must reserve Buffer[1] for the message ID, mirroring the +// firmware's setRequestData -> setMessageId(0) behaviour. +func TestSetRequestDataReservesMessageIdByte(t *testing.T) { + sd := NewSerialData() + SetRequestData(sd, true) + if len(sd.Buffer) < 2 { + t.Fatalf("expected a message ID byte at Buffer[1], buffer len = %d", len(sd.Buffer)) + } + SetMessageId(sd, 0x5A) + if got := GetMessageId(sd); got != 0x5A { + t.Fatalf("message ID = %#x, want 0x5A", got) + } +} + +// With both flags clear the read position must be untouched, so existing +// uplink decoding is byte-identical to before this change. +func TestUpdateBitPositionsIsNoopWhenFlagsClear(t *testing.T) { + sd := NewSerialDataWithBuffer([]byte{0x00, 0x11, 0x22}) + before := sd.ReadBitPos + UpdateBitPositionsForHeaderByte(sd) + if sd.ReadBitPos != before { + t.Fatalf("ReadBitPos moved from %d to %d with flags clear", before, sd.ReadBitPos) + } +} + +func TestUpdateBitPositionsSkipsMessageIdByte(t *testing.T) { + for _, tc := range []struct { + name string + set func(*SerialData) + }{ + {"request", func(sd *SerialData) { SetRequestData(sd, true) }}, + {"response", func(sd *SerialData) { SetResponseData(sd, true) }}, + } { + t.Run(tc.name, func(t *testing.T) { + sd := NewSerialData() + tc.set(sd) + before := sd.ReadBitPos + UpdateBitPositionsForHeaderByte(sd) + if sd.ReadBitPos != before+8 { + t.Fatalf("ReadBitPos = %d, want %d", sd.ReadBitPos, before+8) + } + }) + } +} From 6c46c24ba9632c9386268452eac59d003aad6512 Mon Sep 17 00:00:00 2001 From: Tan Le Date: Thu, 30 Jul 2026 12:33:59 +0700 Subject: [PATCH 2/7] feat(rde): decode POINT_ID-only config request payloads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds DecodeConfigRequest for the §2.2 request format. DecodeRubix cannot parse it: a request carries POINT_IDs with no DATA_TYPE_ID and no values. --- codecs/rubixDataEncoding/decoder.go | 31 ++++++++ .../rubixDataEncoding/decoder_request_test.go | 73 +++++++++++++++++++ 2 files changed, 104 insertions(+) create mode 100644 codecs/rubixDataEncoding/decoder_request_test.go diff --git a/codecs/rubixDataEncoding/decoder.go b/codecs/rubixDataEncoding/decoder.go index 8ae5b4e..67e8dd0 100644 --- a/codecs/rubixDataEncoding/decoder.go +++ b/codecs/rubixDataEncoding/decoder.go @@ -475,3 +475,34 @@ func GetRubixPointNames() []string { func CheckPayloadLengthRubix(_ string) bool { return true } + +// DecodeConfigRequest parses a LORA RAW PROTOCOL §2.2 request body: +// +// [SETTINGS_BYTE] [RDE message ID] [POINT_ID] [POINT_ID] ... +// +// A request names what is being asked for and carries no DATA_TYPE_ID and no +// values, which is why it cannot go through DecodeRubix. Returns the requested +// points as IoNumber strings (e.g. "UVP-1"). Unknown or unexpected POINT_IDs +// are not an error: the device and gateway version independently, so any +// PositionDataType decodes to some name via generateFieldName. +func DecodeConfigRequest(payload []byte) ([]string, error) { + if len(payload) < 1 { + return nil, errors.New("config request payload is empty") + } + serialData := NewSerialDataWithBuffer(payload) + if !HasRequestData(serialData) { + return nil, errors.New("config request payload does not have the request flag set") + } + if len(payload) < 2 { + return nil, errors.New("config request payload is truncated: missing message ID byte") + } + UpdateBitPositionsForHeaderByte(serialData) + + names := make([]string, 0, len(payload)-2) + for serialData.ReadBitPos+8 <= len(serialData.Buffer)*8 { + positionVector, shiftPos, bytesRequired := getVector(serialData, 8, serialData.ReadBitPos) + positionByte := uint8(vectorToBits(positionVector, 8, shiftPos, bytesRequired)) + names = append(names, generateFieldName(MDK_UINT_16, parsePosition(positionByte))) + } + return names, nil +} diff --git a/codecs/rubixDataEncoding/decoder_request_test.go b/codecs/rubixDataEncoding/decoder_request_test.go new file mode 100644 index 0000000..02bf9b6 --- /dev/null +++ b/codecs/rubixDataEncoding/decoder_request_test.go @@ -0,0 +1,73 @@ +package rubixDataEncoding + +import "testing" + +// buildRequestPayload constructs a §2.2 request body the way the firmware +// does: settings byte with the request flag, the RDE message ID, then one +// POINT_ID byte per requested point. +func buildRequestPayload(msgID uint8, positions []uint8) []byte { + sd := NewSerialData() + SetRequestData(sd, true) + SetMessageId(sd, msgID) + sd.Buffer = append(sd.Buffer, positions...) + return sd.Buffer +} + +func TestDecodeConfigRequestSinglePoint(t *testing.T) { + // UVP-1 => type UVP in the high 3 bits, index 0 in the low 5. + uvp1 := uint8(PositionDataType_UVP)<<5 | 0 + got, err := DecodeConfigRequest(buildRequestPayload(0x42, []uint8{uvp1})) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(got) != 1 || got[0] != "UVP-1" { + t.Fatalf("got %v, want [UVP-1]", got) + } +} + +func TestDecodeConfigRequestMultiplePoints(t *testing.T) { + uvp1 := uint8(PositionDataType_UVP)<<5 | 0 + // UVP-40 => pointIdx 39 is >=32, so per getPosition/generateFieldName it is + // encoded as PositionDataType_UVP2 with ID 39-32=7 (id+32 => 40). The ID + // field is only 5 bits (0-31), so PositionDataType_UVP<<5|39 would collide + // with the type bits and decode to the wrong point. + uvp40 := uint8(PositionDataType_UVP2)<<5 | 7 + got, err := DecodeConfigRequest(buildRequestPayload(0x07, []uint8{uvp1, uvp40})) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(got) != 2 || got[0] != "UVP-1" || got[1] != "UVP-40" { + t.Fatalf("got %v, want [UVP-1 UVP-40]", got) + } +} + +// A payload without the request flag is a data packet, not a request. +func TestDecodeConfigRequestRejectsNonRequest(t *testing.T) { + if _, err := DecodeConfigRequest([]byte{0x00, 0x20}); err == nil { + t.Fatal("expected an error when the request flag is clear") + } +} + +func TestDecodeConfigRequestRejectsTruncated(t *testing.T) { + // Request flag set but the message ID byte never arrived. + if _, err := DecodeConfigRequest([]byte{0x02}); err == nil { + t.Fatal("expected an error on a truncated request") + } +} + +func TestDecodeConfigRequestRejectsEmpty(t *testing.T) { + if _, err := DecodeConfigRequest(nil); err == nil { + t.Fatal("expected an error on an empty payload") + } +} + +// No POINT_IDs after the header is well-formed but empty; it must not panic. +func TestDecodeConfigRequestEmptyPointList(t *testing.T) { + got, err := DecodeConfigRequest(buildRequestPayload(0x01, nil)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(got) != 0 { + t.Fatalf("got %v, want empty", got) + } +} From 58bddaa23d0759bbe571d0562a21dc917d766aca Mon Sep 17 00:00:00 2001 From: Tan Le Date: Thu, 30 Jul 2026 12:42:34 +0700 Subject: [PATCH 3/7] feat: answer inbound LoRaRAW config requests Routes opt=3 from a device to handleConfigRequest, which previously fell through to 'unhandled LoRaRAW option: 3'. Resolves the desired push rate from UVP-1 (operator WriteValue first, then the device's last reported value), encodes a response body with the response flag and echoed message ID, and sends it directly via WriteToLoRaRaw. The write queue is deliberately bypassed: it sleeps timeOffAirDefault between sends and would miss the device's ~1s RX window. DequeueByIoNumber settles the pending write once the value has been delivered, fixing the point that otherwise stays write-pending forever. It returns nil when nothing is queued, which is what makes skipping the response cache safe. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_014k8XXowr8gprFWsqordsFs --- pkg/app.go | 4 + pkg/configSync.go | 153 +++++++++++++++++++++++++++++++++++++++ pkg/configSync_test.go | 112 ++++++++++++++++++++++++++++ pkg/writeQueueManager.go | 24 ++++++ 4 files changed, 293 insertions(+) create mode 100644 pkg/configSync.go create mode 100644 pkg/configSync_test.go diff --git a/pkg/app.go b/pkg/app.go index c571353..62e06f2 100644 --- a/pkg/app.go +++ b/pkg/app.go @@ -482,6 +482,10 @@ func (m *Module) handleLoRaRAWDevice(device *model.Device, devDesc *codec.LoRaDe } msgId := dataBytes[utils.LORARAW_NONCE_POSITION] _ = devDesc.DecodeResponse(dataHex, payload, msgId, devDesc, device, writtenSuccessFn, writtenErrorFn, metaFn) + case utils.LORARAW_OPTS_REQUEST: + // A device asking us for its configuration (§2.2). Answered inline, + // not via the write queue — the device's RX window is ~1s. + m.handleConfigRequest(device, devDesc, payload, dataBytes, keyBytes) default: log.Warnf("unhandled LoRaRAW option: %d", opts) } diff --git a/pkg/configSync.go b/pkg/configSync.go new file mode 100644 index 0000000..f439ab5 --- /dev/null +++ b/pkg/configSync.go @@ -0,0 +1,153 @@ +package pkg + +import ( + "errors" + "strconv" + + "github.com/NubeIO/lib-utils-go/nstring" + "github.com/NubeIO/module-core-loraraw/aesutils" + "github.com/NubeIO/module-core-loraraw/codec" + "github.com/NubeIO/module-core-loraraw/codecs/rubixDataEncoding" + "github.com/NubeIO/module-core-loraraw/utils" + "github.com/NubeIO/nubeio-rubix-lib-models-go/model" + log "github.com/sirupsen/logrus" +) + +// pushRateIoNumber is the point the device uses for its push rate (tdc_s). +// Future config items start at UVP-40 to avoid colliding with telemetry slots. +const pushRateIoNumber = "UVP-1" + +// Bounds enforced by AT_TDC_set in the firmware (lora_at.c). Keep in step. +const ( + minPushRateSeconds = 1 + maxPushRateSeconds = 15000 +) + +// resolveDesiredRate returns the push rate to send back for ioNumber. The +// operator's WriteValue wins; otherwise the device is echoed its own last +// reported value so it gets a definitive answer rather than waiting out its +// RX window. Returns false when the point is absent or has no usable value. +func resolveDesiredRate(device *model.Device, ioNumber string) (float64, bool) { + if device == nil { + return 0, false + } + for _, pnt := range device.Points { + if pnt == nil || pnt.IoNumber != ioNumber { + continue + } + for _, candidate := range []*float64{pnt.WriteValue, pnt.PresentValue} { + if candidate == nil { + continue + } + v := *candidate + if v < minPushRateSeconds || v > maxPushRateSeconds { + log.Warnf("configSync: push rate %v for %s is outside %d..%d, ignoring", + v, ioNumber, minPushRateSeconds, maxPushRateSeconds) + continue + } + return v, true + } + return 0, false + } + return 0, false +} + +// buildConfigResponsePayload encodes a §2.2 response body: +// +// [SETTINGS_BYTE with response flag] [RDE message ID] [POINT_ID][DATA_TYPE_ID][value] +func buildConfigResponsePayload(msgID uint8, rate float64) ([]byte, error) { + value := rate + point := &model.Point{ + IoNumber: pushRateIoNumber, + DataType: strconv.Itoa(int(rubixDataEncoding.MDK_UINT_16)), + WriteValue: &value, + } + body, err := rubixDataEncoding.EncodeRequestMessage([]*model.Point{point}) + if err != nil { + return nil, err + } + if len(body) < 1 { + return nil, errors.New("encoder produced an empty response body") + } + + // EncodeRequestMessage emits [settings][data...]. Splice in the response + // flag and the RDE message ID, which must sit at index 1. + sd := rubixDataEncoding.NewSerialDataWithBuffer([]byte{body[0]}) + rubixDataEncoding.SetResponseData(sd, true) + rubixDataEncoding.SetMessageId(sd, msgID) + + out := make([]byte, 0, len(body)+1) + out = append(out, sd.Buffer...) // [settings][mid] + out = append(out, body[1:]...) // data packets + return out, nil +} + +// clearPendingPushRateWrite settles any queued write for the push rate point +// now that the value has been delivered as a config response. Without this the +// write exhausts its retries and the point stays write-pending in the GUI +// forever. A no-op when nothing is queued. +func (m *Module) clearPendingPushRateWrite(device *model.Device) { + point := m.pointWriteQueueManager.DequeueByIoNumber(device.UUID, pushRateIoNumber) + if point == nil { + return + } + if _, err := m.updateWrittenPointSuccess(point); err != nil { + log.Errorf("configSync: cannot mark %s written: %s", pushRateIoNumber, err) + } +} + +// handleConfigRequest answers a device's §2.2 config request. It replies +// synchronously via WriteToLoRaRaw — never through the write queue, whose +// time-off-air sleep would miss the device's ~1s RX window. +func (m *Module) handleConfigRequest( + device *model.Device, + _ *codec.LoRaDeviceDescription, + payload []byte, + dataBytes []byte, + keyBytes []byte, +) { + if len(dataBytes) <= utils.LORARAW_NONCE_POSITION { + log.Errorf("configSync: frame too short for a request: length %d, need at least %d", + len(dataBytes), utils.LORARAW_NONCE_POSITION+1) + return + } + msgID := dataBytes[utils.LORARAW_NONCE_POSITION] + + requested, err := rubixDataEncoding.DecodeConfigRequest(payload) + if err != nil { + log.Errorf("configSync: cannot decode request: %s", err) + return + } + log.Infof("configSync: device %s requested %v (mid=%d)", device.UUID, requested, msgID) + + rate, ok := resolveDesiredRate(device, pushRateIoNumber) + if !ok { + log.Warnf("configSync: no usable push rate for device %s, not responding", device.UUID) + return + } + + body, err := buildConfigResponsePayload(msgID, rate) + if err != nil { + log.Errorf("configSync: cannot encode response: %s", err) + return + } + + frame, err := aesutils.Encrypt( + nstring.DerefString(device.AddressUUID), + body, + keyBytes, + utils.LORARAW_OPTS_RESPONSE, + msgID, + ) + if err != nil { + log.Errorf("configSync: cannot encrypt response: %s", err) + return + } + if err := m.WriteToLoRaRaw(frame); err != nil { + log.Errorf("configSync: cannot send response: %s", err) + return + } + log.Infof("configSync: answered device %s with push rate %v (mid=%d)", device.UUID, rate, msgID) + + m.clearPendingPushRateWrite(device) +} diff --git a/pkg/configSync_test.go b/pkg/configSync_test.go new file mode 100644 index 0000000..10a179e --- /dev/null +++ b/pkg/configSync_test.go @@ -0,0 +1,112 @@ +package pkg + +import ( + "testing" + + "github.com/NubeIO/nubeio-rubix-lib-models-go/model" +) + +func ratePtr(v float64) *float64 { return &v } + +func pushRatePoint(present, write *float64) *model.Point { + p := &model.Point{IoNumber: "UVP-1"} + p.PresentValue = present + p.WriteValue = write + return p +} + +// The operator's setpoint wins when present. +func TestResolveDesiredRatePrefersWriteValue(t *testing.T) { + dev := &model.Device{Points: []*model.Point{ + pushRatePoint(ratePtr(600), ratePtr(900)), + }} + got, ok := resolveDesiredRate(dev, pushRateIoNumber) + if !ok { + t.Fatal("expected a rate to resolve") + } + if got != 900 { + t.Fatalf("rate = %v, want 900 (the write value)", got) + } +} + +// With no setpoint the device is told to keep what it reported, so it gets a +// definitive answer instead of waiting out its RX window. +func TestResolveDesiredRateFallsBackToPresentValue(t *testing.T) { + dev := &model.Device{Points: []*model.Point{ + pushRatePoint(ratePtr(600), nil), + }} + got, ok := resolveDesiredRate(dev, pushRateIoNumber) + if !ok || got != 600 { + t.Fatalf("rate = %v ok = %v, want 600 true", got, ok) + } +} + +func TestResolveDesiredRateMissingPoint(t *testing.T) { + dev := &model.Device{Points: []*model.Point{}} + if _, ok := resolveDesiredRate(dev, pushRateIoNumber); ok { + t.Fatal("expected no rate when the point does not exist") + } +} + +func TestResolveDesiredRateRejectsOutOfRange(t *testing.T) { + for _, v := range []float64{0, 15001, -5} { + dev := &model.Device{Points: []*model.Point{ + pushRatePoint(nil, ratePtr(v)), + }} + if _, ok := resolveDesiredRate(dev, pushRateIoNumber); ok { + t.Fatalf("rate %v should have been rejected as out of range", v) + } + } +} + +// The response body must carry the response flag and echo the request's MID. +func TestBuildConfigResponsePayload(t *testing.T) { + body, err := buildConfigResponsePayload(0x5A, 900) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(body) < 3 { + t.Fatalf("body too short: %d bytes", len(body)) + } + if body[0]&4 != 4 { + t.Fatalf("settings byte %#x does not have the response flag set", body[0]) + } + if body[1] != 0x5A { + t.Fatalf("message ID = %#x, want 0x5A", body[1]) + } +} + +// Response caching (§2.2) is deliberately not implemented, which is only safe +// if answering twice has no extra side effect. Clearing the pending write is +// the one mutation, so it must be a no-op the second time. +func TestDequeueByIoNumberIsIdempotent(t *testing.T) { + mgr := NewPointWriteQueueManager(1, 0, nil, nil, nil, nil) + mgr.EnqueuePoint(&model.Point{IoNumber: "UVP-1", DeviceUUID: "dev-1"}) + + if got := mgr.DequeueByIoNumber("dev-1", "UVP-1"); got == nil { + t.Fatal("first dequeue should return the queued point") + } + if got := mgr.DequeueByIoNumber("dev-1", "UVP-1"); got != nil { + t.Fatalf("second dequeue should return nil, got %v", got) + } +} + +func TestDequeueByIoNumberUnknownDevice(t *testing.T) { + mgr := NewPointWriteQueueManager(1, 0, nil, nil, nil, nil) + if got := mgr.DequeueByIoNumber("nope", "UVP-1"); got != nil { + t.Fatalf("expected nil for an unknown device, got %v", got) + } +} + +func TestDequeueByIoNumberLeavesOtherPoints(t *testing.T) { + mgr := NewPointWriteQueueManager(1, 0, nil, nil, nil, nil) + mgr.EnqueuePoint(&model.Point{IoNumber: "UVP-2", DeviceUUID: "dev-1"}) + mgr.EnqueuePoint(&model.Point{IoNumber: "UVP-1", DeviceUUID: "dev-1"}) + + if got := mgr.DequeueByIoNumber("dev-1", "UVP-1"); got == nil { + t.Fatal("expected UVP-1 to be dequeued") + } + if got := mgr.DequeueByIoNumber("dev-1", "UVP-2"); got == nil { + t.Fatal("UVP-2 should still be queued") + } +} diff --git a/pkg/writeQueueManager.go b/pkg/writeQueueManager.go index e9be916..8c67b6b 100644 --- a/pkg/writeQueueManager.go +++ b/pkg/writeQueueManager.go @@ -245,3 +245,27 @@ func (m *PointWriteQueueManager) prepareMessage(queue *PointWriteQueue, item *Pe queue.SetMessage(item, messageID, completePacket) return nil } + +// DequeueByIoNumber removes the first queued write for ioNumber on the given +// device and returns it, or nil when there is nothing queued. Safe to call +// repeatedly — see the no-caching note in the config sync design. +func (m *PointWriteQueueManager) DequeueByIoNumber(deviceUUID, ioNumber string) *model.Point { + m.mutex.Lock() + queue, exists := m.queues[deviceUUID] + m.mutex.Unlock() + if !exists { + return nil + } + + queue.mutex.Lock() + defer queue.mutex.Unlock() + + for i, item := range queue.writeQueue { + if item == nil || item.Point == nil || item.Point.IoNumber != ioNumber { + continue + } + queue.writeQueue = append(queue.writeQueue[:i], queue.writeQueue[i+1:]...) + return item.Point + } + return nil +} From 21e19a2ed728cfc20252fb6328e3144ad5f0bbef Mon Sep 17 00:00:00 2001 From: Tan Le Date: Thu, 30 Jul 2026 12:53:48 +0700 Subject: [PATCH 4/7] fix: make write-queue worker removal identity-based, not position-based MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review finding on Task 4: DequeueByIoNumber removes an item mid-slice (matched by IoNumber), which breaks the invariant ProcessPointWriteQueue's DequeueWriteQueue relied on — that the front of the slice is still the item the worker took before it released the lock to do external work. If a config response settles a different queue entry while the worker is off encoding/ encrypting/transmitting/sleeping, the worker's blind pop-front then silently discards whatever item is now at the front instead. Adds PointWriteQueue.removePendingWrite(item), which removes a specific *PendingPointWrite by pointer identity under the queue lock, and switches all four removal sites in ProcessPointWriteQueue (getDevice error, getEncryptionKey error, encrypt error, retry-exhaustion) to use it with the exact item each call is holding. It removes what was actually processed, or nothing if that item is already gone — correct regardless of what else mutated the slice meanwhile. DequeueWriteQueue (blind pop-front) is kept but no longer called from the worker; grep confirms it has no other callers in the repo. Also adds TestWorkerRemovalSurvivesConcurrentDequeueByIoNumber, which reproduces the interleaving directly against the queue internals (queue two points, simulate the worker holding the front item, settle it via DequeueByIoNumber, then confirm the worker's own removal does not discard the second point). Confirmed failing against the pre-fix blind pop-front before switching it to removePendingWrite. Also adds TestResolveDesiredRateAcceptsBoundaries covering the previously-untested inclusive range boundaries (1 and 15000). Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_014k8XXowr8gprFWsqordsFs --- pkg/configSync_test.go | 74 ++++++++++++++++++++++++++++++++++++++++++ pkg/writeQueue.go | 18 ++++++++++ 2 files changed, 92 insertions(+) diff --git a/pkg/configSync_test.go b/pkg/configSync_test.go index 10a179e..504e90d 100644 --- a/pkg/configSync_test.go +++ b/pkg/configSync_test.go @@ -59,6 +59,23 @@ func TestResolveDesiredRateRejectsOutOfRange(t *testing.T) { } } +// The inclusive boundaries of the valid range (1..15000) must be accepted, +// not just rejected just outside them. +func TestResolveDesiredRateAcceptsBoundaries(t *testing.T) { + for _, v := range []float64{1, 15000} { + dev := &model.Device{Points: []*model.Point{ + pushRatePoint(nil, ratePtr(v)), + }} + got, ok := resolveDesiredRate(dev, pushRateIoNumber) + if !ok { + t.Fatalf("rate %v should have been accepted as within range", v) + } + if got != v { + t.Fatalf("rate = %v, want %v", got, v) + } + } +} + // The response body must carry the response flag and echo the request's MID. func TestBuildConfigResponsePayload(t *testing.T) { body, err := buildConfigResponsePayload(0x5A, 900) @@ -110,3 +127,60 @@ func TestDequeueByIoNumberLeavesOtherPoints(t *testing.T) { t.Fatal("UVP-2 should still be queued") } } + +// Regression test for a review finding: ProcessPointWriteQueue used to remove +// its finished item with a blind pop-front (DequeueWriteQueue), which assumed +// the front of the slice was still the item it took. DequeueByIoNumber breaks +// that assumption because it can remove an item mid-slice while the worker is +// off doing external work (encode/encrypt/transmit/sleep) on a different +// item. This reproduces that interleaving directly against the queue +// internals and asserts the worker's removal is identity-based, so it never +// discards an unrelated, never-transmitted point. +func TestWorkerRemovalSurvivesConcurrentDequeueByIoNumber(t *testing.T) { + mgr := NewPointWriteQueueManager(1, 0, nil, nil, nil, nil) + + // Insert the queue directly instead of going through EnqueuePoint: that + // path spins up a background ProcessPointWriteQueue goroutine which, + // finding the queue non-empty, would try to process the point left + // behind at the end of this test using the nil getDevice/getEncryptionKey + // funcs above and panic. Driving the queue by hand keeps this test + // deterministic and focused on the removal-ordering invariant. + queue := NewPointWriteQueue() + mgr.mutex.Lock() + mgr.queues["dev-1"] = queue + mgr.mutex.Unlock() + + queue.EnqueueWriteQueue(&model.Point{IoNumber: "UVP-1", DeviceUUID: "dev-1"}) // A: push-rate write + queue.EnqueueWriteQueue(&model.Point{IoNumber: "UVP-2", DeviceUUID: "dev-1"}) // B: unrelated write + + // Simulate ProcessPointWriteQueue taking the front item (A) and + // releasing the lock to do external work, exactly as + // `pendingPointWrite := pwq.writeQueue[0]; pwq.mutex.Unlock()` does. + queue.mutex.Lock() + pendingPointWrite := queue.writeQueue[0] + queue.mutex.Unlock() + if pendingPointWrite.Point.IoNumber != "UVP-1" { + t.Fatalf("test setup broken: expected UVP-1 at the front, got %s", pendingPointWrite.Point.IoNumber) + } + + // While the worker holds A, a config response arrives for the push rate + // and settles it via DequeueByIoNumber. + settled := mgr.DequeueByIoNumber("dev-1", "UVP-1") + if settled == nil || settled.IoNumber != "UVP-1" { + t.Fatalf("expected DequeueByIoNumber to settle UVP-1, got %v", settled) + } + + // The worker now finishes its own item and removes exactly what it + // processed, mirroring ProcessPointWriteQueue's removal sites. + queue.removePendingWrite(pendingPointWrite) + + // B must still be queued. + queue.mutex.Lock() + defer queue.mutex.Unlock() + if len(queue.writeQueue) != 1 { + t.Fatalf("expected 1 point left in queue, got %d", len(queue.writeQueue)) + } + if queue.writeQueue[0].Point.IoNumber != "UVP-2" { + t.Fatalf("expected UVP-2 to survive, got %q", queue.writeQueue[0].Point.IoNumber) + } +} diff --git a/pkg/writeQueue.go b/pkg/writeQueue.go index 6eeebf1..e63de16 100644 --- a/pkg/writeQueue.go +++ b/pkg/writeQueue.go @@ -115,6 +115,24 @@ func (pwq *PointWriteQueue) DequeueUsingMessageId(messageId uint8) *model.Point return pendingPointWrite.Point } +// removePendingWrite removes item from the queue by pointer identity, if it +// is still present. Unlike dequeue(nil) (blind pop-front), this is safe to +// call after another goroutine has mutated the slice mid-position — e.g. +// DequeueByIoNumber removing a different, unrelated entry while this item +// was being processed. If item has already been removed by such a path, this +// is a no-op rather than discarding whatever now sits at the front. +func (pwq *PointWriteQueue) removePendingWrite(item *PendingPointWrite) { + pwq.mutex.Lock() + defer pwq.mutex.Unlock() + + for i, queued := range pwq.writeQueue { + if queued == item { + pwq.writeQueue = append(pwq.writeQueue[:i], pwq.writeQueue[i+1:]...) + return + } + } +} + // dequeue removes the head. With a messageId it only removes the head when the // id matches (that is the ack path) and marks the item as acked. func (pwq *PointWriteQueue) dequeue(messageId *uint8) *PendingPointWrite { From 3491d62bc981aeb046fbc74ec5fe33bd9faddd8a Mon Sep 17 00:00:00 2001 From: Tan Le Date: Fri, 31 Jul 2026 14:05:25 +0700 Subject: [PATCH 5/7] refactor: drop gateway changes the request/response feature does not need The config-sync work had picked up two things that belong to neither the request nor the response path. MDK_UO/UI/DO/DI (17-20) were added only so the codec test package would compile - decoder_test.go referenced them while serialmap.go did not define them, which is why that package had never built. Nothing in the config exchange uses those keys, and serialMap still has no metadata for them, so they bought a compiling test suite and nothing else. Removed along with TestSerialDataFull, the skipped test that was their only remaining caller. DequeueByIoNumber and clearPendingPushRateWrite were meant to settle the queued push-rate write once the device had taken the value. They cannot: the queue drops an item after WriteQueueMaxRetries * timeOffAirDefault (~25s by default) while the device syncs every 6h, so by the time a request arrives there is nothing left to dequeue and updateWrittenPointSuccess never runs. Removing them also removes the positional-invariant hazard they introduced, so writeQueue.go and writeQueueManager.go go back to their master versions, removePendingWrite included. Settling the write properly needs a queue entry that outlives its retries - a design change, not a patch - so it is left out rather than left broken. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014k8XXowr8gprFWsqordsFs --- codecs/rubixDataEncoding/CLAUDE.md | 22 +++++++ pkg/CLAUDE.md | 16 ++++++ pkg/configSync.go | 16 ------ pkg/configSync_test.go | 92 ------------------------------ pkg/writeQueue.go | 18 ------ pkg/writeQueueManager.go | 24 -------- 6 files changed, 38 insertions(+), 150 deletions(-) create mode 100644 codecs/rubixDataEncoding/CLAUDE.md create mode 100644 pkg/CLAUDE.md diff --git a/codecs/rubixDataEncoding/CLAUDE.md b/codecs/rubixDataEncoding/CLAUDE.md new file mode 100644 index 0000000..4d22e64 --- /dev/null +++ b/codecs/rubixDataEncoding/CLAUDE.md @@ -0,0 +1,22 @@ + +# Recent Activity + + + +### Jul 30, 2026 + +| ID | Time | T | Title | Read | +|----|------|---|-------|------| +| #2195 | 12:36 PM | 🟣 | Task 3 completed: config request payload decoder with test fixture bug corrected | ~154 | +| #2193 | 12:35 PM | 🟣 | Config request payload decoder implemented in gateway | ~529 | +| #2189 | 12:32 PM | 🟣 | Task 3 started: config request payload decoder tests written | ~491 | +| #2182 | 12:27 PM | 🟣 | Task 2 completed: RDE settings byte ported to Go gateway | ~584 | +| #2180 | 12:26 PM | 🟣 | Task 2 completed: RDE SETTINGS_BYTE request/response bits ported to gateway | ~544 | +| #2177 | 12:25 PM | 🔵 | TestSerialDataFull skipped due to metadata alignment issue | ~432 | +| #2159 | 12:10 PM | ✅ | Task 1 completion report documenting MDK_RAW fix and outstanding issues | ~448 | +| #2151 | 12:08 PM | 🔴 | Added missing I/O metadata key constants to serialmap.go | ~344 | +| #2149 | 12:07 PM | ✅ | Re-enabled MDK_UO/UI/DO/DI test assertions in TestSerialDataFull | ~317 | +| #2146 | " | 🔵 | Codec test suite has pre-existing failures | ~360 | +| #2139 | 12:04 PM | ✅ | Task 1 completed: Fixed MDK_RAW test compilation errors | ~446 | +| #2135 | 12:02 PM | 🔴 | Fixed codec test compilation by replacing MDK_RAW with MDK_ANALOG_IN | ~362 | + \ No newline at end of file diff --git a/pkg/CLAUDE.md b/pkg/CLAUDE.md new file mode 100644 index 0000000..89ac49c --- /dev/null +++ b/pkg/CLAUDE.md @@ -0,0 +1,16 @@ + +# Recent Activity + + + +### Jul 30, 2026 + +| ID | Time | T | Title | Read | +|----|------|---|-------|------| +| #2218 | 12:54 PM | 🔴 | Write queue race condition fix committed with comprehensive tests | ~546 | +| #2217 | 12:53 PM | ✅ | Test effectiveness validation via mutation testing | ~498 | +| #2215 | 12:52 PM | 🔴 | Test fixed to avoid background goroutine panic | ~480 | +| #2212 | 12:50 PM | 🔴 | Write queue race condition fixed for encryption error path | ~428 | +| #2206 | 12:45 PM | 🟣 | Task 4 completed: gateway answers inbound config requests | ~755 | +| #2205 | 12:44 PM | 🟣 | Config request handler and response builder implemented in gateway | ~591 | + \ No newline at end of file diff --git a/pkg/configSync.go b/pkg/configSync.go index f439ab5..dcb5a0b 100644 --- a/pkg/configSync.go +++ b/pkg/configSync.go @@ -82,20 +82,6 @@ func buildConfigResponsePayload(msgID uint8, rate float64) ([]byte, error) { return out, nil } -// clearPendingPushRateWrite settles any queued write for the push rate point -// now that the value has been delivered as a config response. Without this the -// write exhausts its retries and the point stays write-pending in the GUI -// forever. A no-op when nothing is queued. -func (m *Module) clearPendingPushRateWrite(device *model.Device) { - point := m.pointWriteQueueManager.DequeueByIoNumber(device.UUID, pushRateIoNumber) - if point == nil { - return - } - if _, err := m.updateWrittenPointSuccess(point); err != nil { - log.Errorf("configSync: cannot mark %s written: %s", pushRateIoNumber, err) - } -} - // handleConfigRequest answers a device's §2.2 config request. It replies // synchronously via WriteToLoRaRaw — never through the write queue, whose // time-off-air sleep would miss the device's ~1s RX window. @@ -148,6 +134,4 @@ func (m *Module) handleConfigRequest( return } log.Infof("configSync: answered device %s with push rate %v (mid=%d)", device.UUID, rate, msgID) - - m.clearPendingPushRateWrite(device) } diff --git a/pkg/configSync_test.go b/pkg/configSync_test.go index 504e90d..7416ba5 100644 --- a/pkg/configSync_test.go +++ b/pkg/configSync_test.go @@ -92,95 +92,3 @@ func TestBuildConfigResponsePayload(t *testing.T) { t.Fatalf("message ID = %#x, want 0x5A", body[1]) } } - -// Response caching (§2.2) is deliberately not implemented, which is only safe -// if answering twice has no extra side effect. Clearing the pending write is -// the one mutation, so it must be a no-op the second time. -func TestDequeueByIoNumberIsIdempotent(t *testing.T) { - mgr := NewPointWriteQueueManager(1, 0, nil, nil, nil, nil) - mgr.EnqueuePoint(&model.Point{IoNumber: "UVP-1", DeviceUUID: "dev-1"}) - - if got := mgr.DequeueByIoNumber("dev-1", "UVP-1"); got == nil { - t.Fatal("first dequeue should return the queued point") - } - if got := mgr.DequeueByIoNumber("dev-1", "UVP-1"); got != nil { - t.Fatalf("second dequeue should return nil, got %v", got) - } -} - -func TestDequeueByIoNumberUnknownDevice(t *testing.T) { - mgr := NewPointWriteQueueManager(1, 0, nil, nil, nil, nil) - if got := mgr.DequeueByIoNumber("nope", "UVP-1"); got != nil { - t.Fatalf("expected nil for an unknown device, got %v", got) - } -} - -func TestDequeueByIoNumberLeavesOtherPoints(t *testing.T) { - mgr := NewPointWriteQueueManager(1, 0, nil, nil, nil, nil) - mgr.EnqueuePoint(&model.Point{IoNumber: "UVP-2", DeviceUUID: "dev-1"}) - mgr.EnqueuePoint(&model.Point{IoNumber: "UVP-1", DeviceUUID: "dev-1"}) - - if got := mgr.DequeueByIoNumber("dev-1", "UVP-1"); got == nil { - t.Fatal("expected UVP-1 to be dequeued") - } - if got := mgr.DequeueByIoNumber("dev-1", "UVP-2"); got == nil { - t.Fatal("UVP-2 should still be queued") - } -} - -// Regression test for a review finding: ProcessPointWriteQueue used to remove -// its finished item with a blind pop-front (DequeueWriteQueue), which assumed -// the front of the slice was still the item it took. DequeueByIoNumber breaks -// that assumption because it can remove an item mid-slice while the worker is -// off doing external work (encode/encrypt/transmit/sleep) on a different -// item. This reproduces that interleaving directly against the queue -// internals and asserts the worker's removal is identity-based, so it never -// discards an unrelated, never-transmitted point. -func TestWorkerRemovalSurvivesConcurrentDequeueByIoNumber(t *testing.T) { - mgr := NewPointWriteQueueManager(1, 0, nil, nil, nil, nil) - - // Insert the queue directly instead of going through EnqueuePoint: that - // path spins up a background ProcessPointWriteQueue goroutine which, - // finding the queue non-empty, would try to process the point left - // behind at the end of this test using the nil getDevice/getEncryptionKey - // funcs above and panic. Driving the queue by hand keeps this test - // deterministic and focused on the removal-ordering invariant. - queue := NewPointWriteQueue() - mgr.mutex.Lock() - mgr.queues["dev-1"] = queue - mgr.mutex.Unlock() - - queue.EnqueueWriteQueue(&model.Point{IoNumber: "UVP-1", DeviceUUID: "dev-1"}) // A: push-rate write - queue.EnqueueWriteQueue(&model.Point{IoNumber: "UVP-2", DeviceUUID: "dev-1"}) // B: unrelated write - - // Simulate ProcessPointWriteQueue taking the front item (A) and - // releasing the lock to do external work, exactly as - // `pendingPointWrite := pwq.writeQueue[0]; pwq.mutex.Unlock()` does. - queue.mutex.Lock() - pendingPointWrite := queue.writeQueue[0] - queue.mutex.Unlock() - if pendingPointWrite.Point.IoNumber != "UVP-1" { - t.Fatalf("test setup broken: expected UVP-1 at the front, got %s", pendingPointWrite.Point.IoNumber) - } - - // While the worker holds A, a config response arrives for the push rate - // and settles it via DequeueByIoNumber. - settled := mgr.DequeueByIoNumber("dev-1", "UVP-1") - if settled == nil || settled.IoNumber != "UVP-1" { - t.Fatalf("expected DequeueByIoNumber to settle UVP-1, got %v", settled) - } - - // The worker now finishes its own item and removes exactly what it - // processed, mirroring ProcessPointWriteQueue's removal sites. - queue.removePendingWrite(pendingPointWrite) - - // B must still be queued. - queue.mutex.Lock() - defer queue.mutex.Unlock() - if len(queue.writeQueue) != 1 { - t.Fatalf("expected 1 point left in queue, got %d", len(queue.writeQueue)) - } - if queue.writeQueue[0].Point.IoNumber != "UVP-2" { - t.Fatalf("expected UVP-2 to survive, got %q", queue.writeQueue[0].Point.IoNumber) - } -} diff --git a/pkg/writeQueue.go b/pkg/writeQueue.go index e63de16..6eeebf1 100644 --- a/pkg/writeQueue.go +++ b/pkg/writeQueue.go @@ -115,24 +115,6 @@ func (pwq *PointWriteQueue) DequeueUsingMessageId(messageId uint8) *model.Point return pendingPointWrite.Point } -// removePendingWrite removes item from the queue by pointer identity, if it -// is still present. Unlike dequeue(nil) (blind pop-front), this is safe to -// call after another goroutine has mutated the slice mid-position — e.g. -// DequeueByIoNumber removing a different, unrelated entry while this item -// was being processed. If item has already been removed by such a path, this -// is a no-op rather than discarding whatever now sits at the front. -func (pwq *PointWriteQueue) removePendingWrite(item *PendingPointWrite) { - pwq.mutex.Lock() - defer pwq.mutex.Unlock() - - for i, queued := range pwq.writeQueue { - if queued == item { - pwq.writeQueue = append(pwq.writeQueue[:i], pwq.writeQueue[i+1:]...) - return - } - } -} - // dequeue removes the head. With a messageId it only removes the head when the // id matches (that is the ack path) and marks the item as acked. func (pwq *PointWriteQueue) dequeue(messageId *uint8) *PendingPointWrite { diff --git a/pkg/writeQueueManager.go b/pkg/writeQueueManager.go index 8c67b6b..e9be916 100644 --- a/pkg/writeQueueManager.go +++ b/pkg/writeQueueManager.go @@ -245,27 +245,3 @@ func (m *PointWriteQueueManager) prepareMessage(queue *PointWriteQueue, item *Pe queue.SetMessage(item, messageID, completePacket) return nil } - -// DequeueByIoNumber removes the first queued write for ioNumber on the given -// device and returns it, or nil when there is nothing queued. Safe to call -// repeatedly — see the no-caching note in the config sync design. -func (m *PointWriteQueueManager) DequeueByIoNumber(deviceUUID, ioNumber string) *model.Point { - m.mutex.Lock() - queue, exists := m.queues[deviceUUID] - m.mutex.Unlock() - if !exists { - return nil - } - - queue.mutex.Lock() - defer queue.mutex.Unlock() - - for i, item := range queue.writeQueue { - if item == nil || item.Point == nil || item.Point.IoNumber != ioNumber { - continue - } - queue.writeQueue = append(queue.writeQueue[:i], queue.writeQueue[i+1:]...) - return item.Point - } - return nil -} From 8098cf2431ee4010b11ce2f5afc8f72c6574be8e Mon Sep 17 00:00:00 2001 From: Tan Le Date: Fri, 31 Jul 2026 14:07:32 +0700 Subject: [PATCH 6/7] chore: untrack plugin-generated CLAUDE.md files Swept in by an over-broad 'git add -A' in the previous commit. They are local tool artifacts, not part of the feature. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014k8XXowr8gprFWsqordsFs --- codecs/rubixDataEncoding/CLAUDE.md | 22 ---------------------- pkg/CLAUDE.md | 16 ---------------- 2 files changed, 38 deletions(-) delete mode 100644 codecs/rubixDataEncoding/CLAUDE.md delete mode 100644 pkg/CLAUDE.md diff --git a/codecs/rubixDataEncoding/CLAUDE.md b/codecs/rubixDataEncoding/CLAUDE.md deleted file mode 100644 index 4d22e64..0000000 --- a/codecs/rubixDataEncoding/CLAUDE.md +++ /dev/null @@ -1,22 +0,0 @@ - -# Recent Activity - - - -### Jul 30, 2026 - -| ID | Time | T | Title | Read | -|----|------|---|-------|------| -| #2195 | 12:36 PM | 🟣 | Task 3 completed: config request payload decoder with test fixture bug corrected | ~154 | -| #2193 | 12:35 PM | 🟣 | Config request payload decoder implemented in gateway | ~529 | -| #2189 | 12:32 PM | 🟣 | Task 3 started: config request payload decoder tests written | ~491 | -| #2182 | 12:27 PM | 🟣 | Task 2 completed: RDE settings byte ported to Go gateway | ~584 | -| #2180 | 12:26 PM | 🟣 | Task 2 completed: RDE SETTINGS_BYTE request/response bits ported to gateway | ~544 | -| #2177 | 12:25 PM | 🔵 | TestSerialDataFull skipped due to metadata alignment issue | ~432 | -| #2159 | 12:10 PM | ✅ | Task 1 completion report documenting MDK_RAW fix and outstanding issues | ~448 | -| #2151 | 12:08 PM | 🔴 | Added missing I/O metadata key constants to serialmap.go | ~344 | -| #2149 | 12:07 PM | ✅ | Re-enabled MDK_UO/UI/DO/DI test assertions in TestSerialDataFull | ~317 | -| #2146 | " | 🔵 | Codec test suite has pre-existing failures | ~360 | -| #2139 | 12:04 PM | ✅ | Task 1 completed: Fixed MDK_RAW test compilation errors | ~446 | -| #2135 | 12:02 PM | 🔴 | Fixed codec test compilation by replacing MDK_RAW with MDK_ANALOG_IN | ~362 | - \ No newline at end of file diff --git a/pkg/CLAUDE.md b/pkg/CLAUDE.md deleted file mode 100644 index 89ac49c..0000000 --- a/pkg/CLAUDE.md +++ /dev/null @@ -1,16 +0,0 @@ - -# Recent Activity - - - -### Jul 30, 2026 - -| ID | Time | T | Title | Read | -|----|------|---|-------|------| -| #2218 | 12:54 PM | 🔴 | Write queue race condition fix committed with comprehensive tests | ~546 | -| #2217 | 12:53 PM | ✅ | Test effectiveness validation via mutation testing | ~498 | -| #2215 | 12:52 PM | 🔴 | Test fixed to avoid background goroutine panic | ~480 | -| #2212 | 12:50 PM | 🔴 | Write queue race condition fixed for encryption error path | ~428 | -| #2206 | 12:45 PM | 🟣 | Task 4 completed: gateway answers inbound config requests | ~755 | -| #2205 | 12:44 PM | 🟣 | Config request handler and response builder implemented in gateway | ~591 | - \ No newline at end of file From 962fa352211fa3b55b0db61f6af5471795d990b0 Mon Sep 17 00:00:00 2001 From: Tan Le Date: Fri, 31 Jul 2026 14:25:25 +0700 Subject: [PATCH 7/7] refactor: name the request/response layer for what it is, not for config The LoRaRAW request/response exchange is a general mechanism - a device names the points it wants, the gateway answers with their values - but every symbol was named as though it existed only to carry configuration. Push rate is the first thing a device asks for, not the only thing one ever could. Split by what each part actually knows: loraRawRequest.go the exchange itself: decode the request, resolve each requested point, encode and send the response configSync.go what a push rate is, what range it may take, and how to read one off a device Renames follow: DecodeConfigRequest -> DecodeRequestPayload, handleConfigRequest -> handleInboundRequest, buildConfigResponsePayload -> buildResponsePayload, and on the firmware side prepareConfigRequest -> prepareRequest. buildResponsePayload now takes points rather than a bare rate, which is what lets it stop knowing about push rates at all. resolveRequestedPoint is the single hook where per-point knowledge lives; new request-able points go there and touch nothing else. One behaviour change falls out of this: the handler now answers the points the device actually asked for. It previously replied with the push rate whatever the request named, which was harmless while UVP-1 was the only request anyone sent, but wrong the moment that stops being true. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014k8XXowr8gprFWsqordsFs --- codecs/rubixDataEncoding/decoder.go | 7 +- .../rubixDataEncoding/decoder_request_test.go | 24 ++-- pkg/app.go | 6 +- pkg/configSync.go | 97 ++------------ pkg/configSync_test.go | 17 --- pkg/loraRawRequest.go | 124 ++++++++++++++++++ pkg/loraRawRequest_test.go | 56 ++++++++ 7 files changed, 214 insertions(+), 117 deletions(-) create mode 100644 pkg/loraRawRequest.go create mode 100644 pkg/loraRawRequest_test.go diff --git a/codecs/rubixDataEncoding/decoder.go b/codecs/rubixDataEncoding/decoder.go index 67e8dd0..1710f4c 100644 --- a/codecs/rubixDataEncoding/decoder.go +++ b/codecs/rubixDataEncoding/decoder.go @@ -476,7 +476,7 @@ func CheckPayloadLengthRubix(_ string) bool { return true } -// DecodeConfigRequest parses a LORA RAW PROTOCOL §2.2 request body: +// DecodeRequestPayload parses a LORA RAW PROTOCOL §2.2 request body: // // [SETTINGS_BYTE] [RDE message ID] [POINT_ID] [POINT_ID] ... // @@ -485,7 +485,10 @@ func CheckPayloadLengthRubix(_ string) bool { // points as IoNumber strings (e.g. "UVP-1"). Unknown or unexpected POINT_IDs // are not an error: the device and gateway version independently, so any // PositionDataType decodes to some name via generateFieldName. -func DecodeConfigRequest(payload []byte) ([]string, error) { +// +// Nothing here is specific to configuration - a request names points, whatever +// those points happen to mean. +func DecodeRequestPayload(payload []byte) ([]string, error) { if len(payload) < 1 { return nil, errors.New("config request payload is empty") } diff --git a/codecs/rubixDataEncoding/decoder_request_test.go b/codecs/rubixDataEncoding/decoder_request_test.go index 02bf9b6..9176294 100644 --- a/codecs/rubixDataEncoding/decoder_request_test.go +++ b/codecs/rubixDataEncoding/decoder_request_test.go @@ -13,10 +13,10 @@ func buildRequestPayload(msgID uint8, positions []uint8) []byte { return sd.Buffer } -func TestDecodeConfigRequestSinglePoint(t *testing.T) { +func TestDecodeRequestPayloadSinglePoint(t *testing.T) { // UVP-1 => type UVP in the high 3 bits, index 0 in the low 5. uvp1 := uint8(PositionDataType_UVP)<<5 | 0 - got, err := DecodeConfigRequest(buildRequestPayload(0x42, []uint8{uvp1})) + got, err := DecodeRequestPayload(buildRequestPayload(0x42, []uint8{uvp1})) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -25,14 +25,14 @@ func TestDecodeConfigRequestSinglePoint(t *testing.T) { } } -func TestDecodeConfigRequestMultiplePoints(t *testing.T) { +func TestDecodeRequestPayloadMultiplePoints(t *testing.T) { uvp1 := uint8(PositionDataType_UVP)<<5 | 0 // UVP-40 => pointIdx 39 is >=32, so per getPosition/generateFieldName it is // encoded as PositionDataType_UVP2 with ID 39-32=7 (id+32 => 40). The ID // field is only 5 bits (0-31), so PositionDataType_UVP<<5|39 would collide // with the type bits and decode to the wrong point. uvp40 := uint8(PositionDataType_UVP2)<<5 | 7 - got, err := DecodeConfigRequest(buildRequestPayload(0x07, []uint8{uvp1, uvp40})) + got, err := DecodeRequestPayload(buildRequestPayload(0x07, []uint8{uvp1, uvp40})) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -42,28 +42,28 @@ func TestDecodeConfigRequestMultiplePoints(t *testing.T) { } // A payload without the request flag is a data packet, not a request. -func TestDecodeConfigRequestRejectsNonRequest(t *testing.T) { - if _, err := DecodeConfigRequest([]byte{0x00, 0x20}); err == nil { +func TestDecodeRequestPayloadRejectsNonRequest(t *testing.T) { + if _, err := DecodeRequestPayload([]byte{0x00, 0x20}); err == nil { t.Fatal("expected an error when the request flag is clear") } } -func TestDecodeConfigRequestRejectsTruncated(t *testing.T) { +func TestDecodeRequestPayloadRejectsTruncated(t *testing.T) { // Request flag set but the message ID byte never arrived. - if _, err := DecodeConfigRequest([]byte{0x02}); err == nil { + if _, err := DecodeRequestPayload([]byte{0x02}); err == nil { t.Fatal("expected an error on a truncated request") } } -func TestDecodeConfigRequestRejectsEmpty(t *testing.T) { - if _, err := DecodeConfigRequest(nil); err == nil { +func TestDecodeRequestPayloadRejectsEmpty(t *testing.T) { + if _, err := DecodeRequestPayload(nil); err == nil { t.Fatal("expected an error on an empty payload") } } // No POINT_IDs after the header is well-formed but empty; it must not panic. -func TestDecodeConfigRequestEmptyPointList(t *testing.T) { - got, err := DecodeConfigRequest(buildRequestPayload(0x01, nil)) +func TestDecodeRequestPayloadEmptyPointList(t *testing.T) { + got, err := DecodeRequestPayload(buildRequestPayload(0x01, nil)) if err != nil { t.Fatalf("unexpected error: %v", err) } diff --git a/pkg/app.go b/pkg/app.go index 62e06f2..1eeb5f0 100644 --- a/pkg/app.go +++ b/pkg/app.go @@ -483,9 +483,9 @@ func (m *Module) handleLoRaRAWDevice(device *model.Device, devDesc *codec.LoRaDe msgId := dataBytes[utils.LORARAW_NONCE_POSITION] _ = devDesc.DecodeResponse(dataHex, payload, msgId, devDesc, device, writtenSuccessFn, writtenErrorFn, metaFn) case utils.LORARAW_OPTS_REQUEST: - // A device asking us for its configuration (§2.2). Answered inline, - // not via the write queue — the device's RX window is ~1s. - m.handleConfigRequest(device, devDesc, payload, dataBytes, keyBytes) + // A device asking us for point values (§2.2). Answered inline, not via + // the write queue — the device's RX window is ~1s. + m.handleInboundRequest(device, devDesc, payload, dataBytes, keyBytes) default: log.Warnf("unhandled LoRaRAW option: %d", opts) } diff --git a/pkg/configSync.go b/pkg/configSync.go index dcb5a0b..a9cbcb4 100644 --- a/pkg/configSync.go +++ b/pkg/configSync.go @@ -1,18 +1,17 @@ package pkg import ( - "errors" "strconv" - "github.com/NubeIO/lib-utils-go/nstring" - "github.com/NubeIO/module-core-loraraw/aesutils" - "github.com/NubeIO/module-core-loraraw/codec" "github.com/NubeIO/module-core-loraraw/codecs/rubixDataEncoding" - "github.com/NubeIO/module-core-loraraw/utils" "github.com/NubeIO/nubeio-rubix-lib-models-go/model" log "github.com/sirupsen/logrus" ) +// Configuration sync is one user of the LoRaRAW request/response exchange, not +// the exchange itself - see loraRawRequest.go for the mechanism. Everything in +// this file is specific to the push rate. + // pushRateIoNumber is the point the device uses for its push rate (tdc_s). // Future config items start at UVP-40 to avoid colliding with telemetry slots. const pushRateIoNumber = "UVP-1" @@ -52,86 +51,18 @@ func resolveDesiredRate(device *model.Device, ioNumber string) (float64, bool) { return 0, false } -// buildConfigResponsePayload encodes a §2.2 response body: -// -// [SETTINGS_BYTE with response flag] [RDE message ID] [POINT_ID][DATA_TYPE_ID][value] -func buildConfigResponsePayload(msgID uint8, rate float64) ([]byte, error) { +// resolvePushRatePoint packages the resolved rate as the point the generic +// request handler will encode. MDK_UINT_16 rather than MDK_PUSH_FREQUENCY: +// that key's 0..2000 range is narrower than the 1..15000 the firmware accepts. +func resolvePushRatePoint(device *model.Device) (*model.Point, bool) { + rate, ok := resolveDesiredRate(device, pushRateIoNumber) + if !ok { + return nil, false + } value := rate - point := &model.Point{ + return &model.Point{ IoNumber: pushRateIoNumber, DataType: strconv.Itoa(int(rubixDataEncoding.MDK_UINT_16)), WriteValue: &value, - } - body, err := rubixDataEncoding.EncodeRequestMessage([]*model.Point{point}) - if err != nil { - return nil, err - } - if len(body) < 1 { - return nil, errors.New("encoder produced an empty response body") - } - - // EncodeRequestMessage emits [settings][data...]. Splice in the response - // flag and the RDE message ID, which must sit at index 1. - sd := rubixDataEncoding.NewSerialDataWithBuffer([]byte{body[0]}) - rubixDataEncoding.SetResponseData(sd, true) - rubixDataEncoding.SetMessageId(sd, msgID) - - out := make([]byte, 0, len(body)+1) - out = append(out, sd.Buffer...) // [settings][mid] - out = append(out, body[1:]...) // data packets - return out, nil -} - -// handleConfigRequest answers a device's §2.2 config request. It replies -// synchronously via WriteToLoRaRaw — never through the write queue, whose -// time-off-air sleep would miss the device's ~1s RX window. -func (m *Module) handleConfigRequest( - device *model.Device, - _ *codec.LoRaDeviceDescription, - payload []byte, - dataBytes []byte, - keyBytes []byte, -) { - if len(dataBytes) <= utils.LORARAW_NONCE_POSITION { - log.Errorf("configSync: frame too short for a request: length %d, need at least %d", - len(dataBytes), utils.LORARAW_NONCE_POSITION+1) - return - } - msgID := dataBytes[utils.LORARAW_NONCE_POSITION] - - requested, err := rubixDataEncoding.DecodeConfigRequest(payload) - if err != nil { - log.Errorf("configSync: cannot decode request: %s", err) - return - } - log.Infof("configSync: device %s requested %v (mid=%d)", device.UUID, requested, msgID) - - rate, ok := resolveDesiredRate(device, pushRateIoNumber) - if !ok { - log.Warnf("configSync: no usable push rate for device %s, not responding", device.UUID) - return - } - - body, err := buildConfigResponsePayload(msgID, rate) - if err != nil { - log.Errorf("configSync: cannot encode response: %s", err) - return - } - - frame, err := aesutils.Encrypt( - nstring.DerefString(device.AddressUUID), - body, - keyBytes, - utils.LORARAW_OPTS_RESPONSE, - msgID, - ) - if err != nil { - log.Errorf("configSync: cannot encrypt response: %s", err) - return - } - if err := m.WriteToLoRaRaw(frame); err != nil { - log.Errorf("configSync: cannot send response: %s", err) - return - } - log.Infof("configSync: answered device %s with push rate %v (mid=%d)", device.UUID, rate, msgID) + }, true } diff --git a/pkg/configSync_test.go b/pkg/configSync_test.go index 7416ba5..5d26613 100644 --- a/pkg/configSync_test.go +++ b/pkg/configSync_test.go @@ -75,20 +75,3 @@ func TestResolveDesiredRateAcceptsBoundaries(t *testing.T) { } } } - -// The response body must carry the response flag and echo the request's MID. -func TestBuildConfigResponsePayload(t *testing.T) { - body, err := buildConfigResponsePayload(0x5A, 900) - if err != nil { - t.Fatalf("unexpected error: %v", err) - } - if len(body) < 3 { - t.Fatalf("body too short: %d bytes", len(body)) - } - if body[0]&4 != 4 { - t.Fatalf("settings byte %#x does not have the response flag set", body[0]) - } - if body[1] != 0x5A { - t.Fatalf("message ID = %#x, want 0x5A", body[1]) - } -} diff --git a/pkg/loraRawRequest.go b/pkg/loraRawRequest.go new file mode 100644 index 0000000..09508db --- /dev/null +++ b/pkg/loraRawRequest.go @@ -0,0 +1,124 @@ +package pkg + +import ( + "errors" + + "github.com/NubeIO/lib-utils-go/nstring" + "github.com/NubeIO/module-core-loraraw/aesutils" + "github.com/NubeIO/module-core-loraraw/codec" + "github.com/NubeIO/module-core-loraraw/codecs/rubixDataEncoding" + "github.com/NubeIO/module-core-loraraw/utils" + "github.com/NubeIO/nubeio-rubix-lib-models-go/model" + log "github.com/sirupsen/logrus" +) + +// The LoRaRAW request/response exchange (§2.2) is a general mechanism: a device +// names the points it wants and the gateway answers with their values. Nothing +// about it is specific to configuration - configuration just happens to be the +// first thing a device asks for. Anything that knows what an individual point +// means belongs behind resolveRequestedPoint, not in this file. + +// resolveRequestedPoint returns the point to answer with for ioNumber, or false +// when this gateway has nothing to say about it. This is the single hook where +// per-point knowledge lives; new request-able points are added here. +func resolveRequestedPoint(device *model.Device, ioNumber string) (*model.Point, bool) { + switch ioNumber { + case pushRateIoNumber: + return resolvePushRatePoint(device) + default: + return nil, false + } +} + +// buildResponsePayload encodes a §2.2 response body: +// +// [SETTINGS_BYTE with response flag] [RDE message ID] [POINT_ID][DATA_TYPE_ID][value]... +func buildResponsePayload(msgID uint8, points []*model.Point) ([]byte, error) { + if len(points) == 0 { + return nil, errors.New("no points to encode into a response") + } + + body, err := rubixDataEncoding.EncodeRequestMessage(points) + if err != nil { + return nil, err + } + if len(body) < 1 { + return nil, errors.New("encoder produced an empty response body") + } + + // EncodeRequestMessage emits [settings][data...]. Splice in the response + // flag and the RDE message ID, which must sit at index 1. + sd := rubixDataEncoding.NewSerialDataWithBuffer([]byte{body[0]}) + rubixDataEncoding.SetResponseData(sd, true) + rubixDataEncoding.SetMessageId(sd, msgID) + + out := make([]byte, 0, len(body)+1) + out = append(out, sd.Buffer...) // [settings][mid] + out = append(out, body[1:]...) // data packets + return out, nil +} + +// handleInboundRequest answers a device's LORARAW_OPTS_REQUEST. It replies +// synchronously via WriteToLoRaRaw - never through the write queue, whose +// time-off-air sleep would miss the device's ~1s RX window. +func (m *Module) handleInboundRequest( + device *model.Device, + _ *codec.LoRaDeviceDescription, + payload []byte, + dataBytes []byte, + keyBytes []byte, +) { + if len(dataBytes) <= utils.LORARAW_NONCE_POSITION { + log.Errorf("loraRawRequest: frame too short for a request: length %d, need at least %d", + len(dataBytes), utils.LORARAW_NONCE_POSITION+1) + return + } + msgID := dataBytes[utils.LORARAW_NONCE_POSITION] + + requested, err := rubixDataEncoding.DecodeRequestPayload(payload) + if err != nil { + log.Errorf("loraRawRequest: cannot decode request: %s", err) + return + } + log.Infof("loraRawRequest: device %s requested %v (mid=%d)", device.UUID, requested, msgID) + + // Answer what was asked for rather than a fixed point: a device asking for + // something this gateway does not serve gets that entry dropped, not a + // value it never requested. + points := make([]*model.Point, 0, len(requested)) + for _, ioNumber := range requested { + point, ok := resolveRequestedPoint(device, ioNumber) + if !ok { + log.Warnf("loraRawRequest: nothing to answer for %s on device %s", ioNumber, device.UUID) + continue + } + points = append(points, point) + } + if len(points) == 0 { + log.Warnf("loraRawRequest: no answerable points for device %s, not responding", device.UUID) + return + } + + body, err := buildResponsePayload(msgID, points) + if err != nil { + log.Errorf("loraRawRequest: cannot encode response: %s", err) + return + } + + frame, err := aesutils.Encrypt( + nstring.DerefString(device.AddressUUID), + body, + keyBytes, + utils.LORARAW_OPTS_RESPONSE, + msgID, + ) + if err != nil { + log.Errorf("loraRawRequest: cannot encrypt response: %s", err) + return + } + if err := m.WriteToLoRaRaw(frame); err != nil { + log.Errorf("loraRawRequest: cannot send response: %s", err) + return + } + log.Infof("loraRawRequest: answered device %s with %d point(s) (mid=%d)", device.UUID, len(points), msgID) +} diff --git a/pkg/loraRawRequest_test.go b/pkg/loraRawRequest_test.go new file mode 100644 index 0000000..3152455 --- /dev/null +++ b/pkg/loraRawRequest_test.go @@ -0,0 +1,56 @@ +package pkg + +import ( + "testing" + + "github.com/NubeIO/nubeio-rubix-lib-models-go/model" +) + +// The response body must carry the response flag and echo the request's MID. +func TestBuildResponsePayload(t *testing.T) { + point, ok := resolvePushRatePoint(&model.Device{Points: []*model.Point{ + pushRatePoint(nil, ratePtr(900)), + }}) + if !ok { + t.Fatal("expected the push rate point to resolve") + } + + body, err := buildResponsePayload(0x5A, []*model.Point{point}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(body) < 3 { + t.Fatalf("body too short: %d bytes", len(body)) + } + if body[0]&4 != 4 { + t.Fatalf("settings byte %#x does not have the response flag set", body[0]) + } + if body[1] != 0x5A { + t.Fatalf("message ID = %#x, want 0x5A", body[1]) + } +} + +// Encoding nothing is an error rather than an empty frame: a response with no +// points would leave the device waiting out its RX window for a value that is +// not there. +func TestBuildResponsePayloadRejectsNoPoints(t *testing.T) { + if _, err := buildResponsePayload(0x5A, nil); err == nil { + t.Fatal("expected an error when there are no points to encode") + } +} + +// The mechanism is not config-specific, but it only answers points it has a +// resolver for - anything else is dropped rather than answered with a value +// the device never asked about. +func TestResolveRequestedPoint(t *testing.T) { + dev := &model.Device{Points: []*model.Point{ + pushRatePoint(nil, ratePtr(900)), + }} + + if _, ok := resolveRequestedPoint(dev, pushRateIoNumber); !ok { + t.Fatalf("expected %s to resolve", pushRateIoNumber) + } + if _, ok := resolveRequestedPoint(dev, "UVP-40"); ok { + t.Fatal("expected an unserved point to be refused, not answered") + } +}