Release/1.3.5 rc.2

CHANGELOG.md

@@ -1,5 +1,15 @@
 # CHANGELOG
 
+## [v1.3.5-rc.2](https://github.com/NubeIO/module-core-loraraw/tree/v1.3.5-rc.2) (2026-08-18)
+
+- Serialise LoRaRAW writes through a single scheduler: one frame on air at a time, wait for the device RESPONSE (or `write_response_timeout`, default 5s) before the next device's frame, round-robin across devices — concurrent writes to several devices no longer clobber each other's acks on the half-duplex radio
+- Mark a point `api-write-failed` with a message once all write attempts get no response, instead of dropping it silently
+- Count serial write errors as attempts and report the port as disconnected instead of pretending the frame was sent
+- Replace the never-configurable `time_off_air_default` with `write_response_timeout`
+- Fix writes failing with `write queue full` after a plugin disable/enable cycle (serial drainer now restarts on enable)
+- Fix Rubix decoding of `ai-raw` (type 16): its serial map entry was lost in the codec refactor, so every field after it in a frame decoded as garbage
+- Fix `rubixDataEncoding` decoder tests not compiling since the codec refactor
+
 ## [v1.3.4](https://github.com/NubeIO/module-core-loraraw/tree/v1.3.4) (2026-07-27)
 
 - Add encryption-only `RubixEncrypted` model and drop corrupted LoRaRAW frames instead of decoding them as plaintext garbage points

codecs/rubixDataEncoding/decoder_test.go

@@ -6,8 +6,7 @@ import (
 )
 
 func testHeader(t *testing.T, serialData *SerialData, expectedHeader MetaDataKey) {
-	var pos uint8
-	header := getMetaDataKey(serialData, &pos)
+	header, _ := parseMetaData(serialData)
 	if header != expectedHeader {
 		t.Fatalf("Expected header %v, but got %v", expectedHeader, header)
 	}
@@ -143,7 +142,7 @@ func TestDecodefromStm32Encode(t *testing.T) {
 	testDecodeFloat(t, &serialData, 333, MDK_CO2)
 	testDecodeFloat(t, &serialData, 2.9, MDK_BATTERY_VOLTAGE)
 	testDecodeFloat(t, &serialData, 15, MDK_PUSH_FREQUENCY)
-	testDecodeFloat(t, &serialData, 0.888, MDK_RAW)
+	testDecodeFloat(t, &serialData, 0.888, MDK_ANALOG_IN)
 	testDecodeFloat(t, &serialData, 22, MDK_FIRMWARE_VERSION)
 	testDecodeFloat(t, &serialData, 44, MDK_HARDWARE_VERSION)
 	testDecodeUint(t, &serialData, (uint8)(1), MDK_UINT_8)
@@ -202,17 +201,17 @@ func TestSerialDataRaw(t *testing.T) {
 
 	Require(t, canDecode(serialData))
 
-	testDecodeFloat(t, serialData, 0.1, MDK_RAW)
-	testDecodeFloat(t, serialData, 0.2, MDK_RAW)
-	testDecodeFloat(t, serialData, 0.3, MDK_RAW)
-	testDecodeFloat(t, serialData, 0.4, MDK_RAW)
-	testDecodeFloat(t, serialData, 0.5, MDK_RAW)
-	testDecodeFloat(t, serialData, 0.1, MDK_RAW)
-	testDecodeFloat(t, serialData, 0.2, MDK_RAW)
-	testDecodeFloat(t, serialData, 0.3, MDK_RAW)
-	testDecodeFloat(t, serialData, 0.4, MDK_RAW)
-	testDecodeFloat(t, serialData, 0.5, MDK_RAW)
-	testDecodeFloat(t, serialData, 0.6, MDK_RAW)
+	testDecodeFloat(t, serialData, 0.1, MDK_ANALOG_IN)
+	testDecodeFloat(t, serialData, 0.2, MDK_ANALOG_IN)
+	testDecodeFloat(t, serialData, 0.3, MDK_ANALOG_IN)
+	testDecodeFloat(t, serialData, 0.4, MDK_ANALOG_IN)
+	testDecodeFloat(t, serialData, 0.5, MDK_ANALOG_IN)
+	testDecodeFloat(t, serialData, 0.1, MDK_ANALOG_IN)
+	testDecodeFloat(t, serialData, 0.2, MDK_ANALOG_IN)
+	testDecodeFloat(t, serialData, 0.3, MDK_ANALOG_IN)
+	testDecodeFloat(t, serialData, 0.4, MDK_ANALOG_IN)
+	testDecodeFloat(t, serialData, 0.5, MDK_ANALOG_IN)
+	testDecodeFloat(t, serialData, 0.6, MDK_ANALOG_IN)
 	testDecodeFloat(t, serialData, 1, MDK_DIGITAL)
 	testDecodeFloat(t, serialData, 1, MDK_DIGITAL)
 
@@ -246,42 +245,9 @@ func TestPositional(t *testing.T) {
 
 	Require(t, canDecode(serialData))
 
-	testDecodeFloat(t, serialData, 0.1, MDK_RAW)
-	testDecodeFloat(t, serialData, 0.3, MDK_RAW)
-	testDecodeFloat(t, serialData, 0.5, MDK_RAW)
-	testDecodeFloat(t, serialData, 0.7, MDK_RAW)
-	requireFalse(t, canDecode(serialData), "Should not decode")
-}
-
-func TestSerialDataFull(t *testing.T) {
-	pl := []byte{0, 5, 92, 240, 74, 217, 134, 205, 44, 36, 83, 13, 63, 26,
-		62, 240, 68, 192, 41, 178, 7, 11, 166, 152, 233, 160, 61, 13,
-		225, 17, 145, 35, 33, 50, 159, 69, 190, 44}
-	serialData := NewSerialDataWithBuffer(pl)
-
-	Require(t, canDecode(serialData))
-
-	testDecodeFloat(t, serialData, 66.66, MDK_TEMP)
-	testDecodeFloat(t, serialData, 55.55, MDK_RH)
-	testDecodeFloat(t, serialData, 26262, MDK_LUX)
-	testDecodeFloat(t, serialData, 1, MDK_MOVEMENT)
-	testDecodeFloat(t, serialData, 199999, MDK_COUNTER)
-	testDecodeFloat(t, serialData, 1, MDK_DIGITAL)
-	testDecodeFloat(t, serialData, 8.88, MDK_VOLTAGE_0_10)
-	testDecodeFloat(t, serialData, 16.16, MDK_MILLIAMPS_4_20)
-	testDecodeFloat(t, serialData, 444444, MDK_OHM)
-	testDecodeFloat(t, serialData, 333, MDK_CO2)
-	testDecodeFloat(t, serialData, 2.9, MDK_BATTERY_VOLTAGE)
-	testDecodeFloat(t, serialData, 15, MDK_PUSH_FREQUENCY)
-	testDecodeFloat(t, serialData, 0.888, MDK_RAW)
-
-	testDecodeFloat(t, serialData, 0.1, MDK_UO)
-	testDecodeFloat(t, serialData, 0.2, MDK_UI)
-	testDecodeFloat(t, serialData, 0, MDK_DO)
-	testDecodeFloat(t, serialData, 1, MDK_DI)
-
-	testDecodeFloat(t, serialData, 22, MDK_FIRMWARE_VERSION)
-	testDecodeFloat(t, serialData, 44, MDK_HARDWARE_VERSION)
-
+	testDecodeFloat(t, serialData, 0.1, MDK_ANALOG_IN)
+	testDecodeFloat(t, serialData, 0.3, MDK_ANALOG_IN)
+	testDecodeFloat(t, serialData, 0.5, MDK_ANALOG_IN)
+	testDecodeFloat(t, serialData, 0.7, MDK_ANALOG_IN)
 	requireFalse(t, canDecode(serialData), "Should not decode")
 }

codecs/rubixDataEncoding/serialmap.go

@@ -64,6 +64,7 @@ var serialMap = map[MetaDataKey]MetaData{
 	MDK_CO2:              {FIXEDPOINT, 0, 400, 0, 0},
 	MDK_BATTERY_VOLTAGE:  {FIXEDPOINT, 0, 6, 1, 0},
 	MDK_PUSH_FREQUENCY:   {FIXEDPOINT, 0, 2000, 0, 0},
+	MDK_ANALOG_IN:        {FIXEDPOINT, 0, 1, 3, 0},
 	MDK_FIRMWARE_VERSION: {FIXEDPOINT, 0, 255, 0, 0},
 	MDK_HARDWARE_VERSION: {FIXEDPOINT, 0, 255, 0, 0},
 	MDK_UINT_8:           {DATAPOINT, 0, 0, 0, 1},

pkg/app.go

@@ -742,27 +742,56 @@ func (m *Module) getEncryptionKey(device *model.Device) ([]byte, error) {
 	return key, nil
 }
 
+// initWriteQueue starts the serial drainer if it is not running. It is safe to
+// call after stopWriteQueue, so an Enable following a Disable gets a fresh queue.
 func (m *Module) initWriteQueue() {
-	m.writeQueueInit.Do(func() {
-		m.writeQueue = make(chan []byte, 100)
-		m.writeQueueDone = make(chan struct{})
+	m.writeQueueMutex.Lock()
+	defer m.writeQueueMutex.Unlock()
 
-		go m.processWriteQueue()
-	})
+	if m.writeQueue != nil {
+		return
+	}
+	m.writeQueue = make(chan []byte, 100)
+	m.writeQueueDone = make(chan struct{})
+
+	go m.processWriteQueue(m.writeQueue, m.writeQueueDone)
+}
+
+// stopWriteQueue stops the serial drainer. The send channel is never closed:
+// a concurrent WriteToLoRaRaw may still be holding it, and sending on a
+// closed channel panics. Dropping the reference is enough.
+func (m *Module) stopWriteQueue() {
+	m.writeQueueMutex.Lock()
+	defer m.writeQueueMutex.Unlock()
+
+	if m.writeQueue == nil {
+		return
+	}
+	close(m.writeQueueDone)
+	m.writeQueue = nil
+	m.writeQueueDone = nil
+}
+
+// getWriteQueue returns the current send channel, or nil when stopped.
+func (m *Module) getWriteQueue() chan []byte {
+	m.writeQueueMutex.Lock()
+	defer m.writeQueueMutex.Unlock()
+
+	return m.writeQueue
 }
 
-func (m *Module) processWriteQueue() {
+func (m *Module) processWriteQueue(queue <-chan []byte, done <-chan struct{}) {
 	defer func() {
 		if r := recover(); r != nil {
 			log.Errorf("Recovered panic in processWriteQueue: %v", r)
 			// Restart goroutine
-			go m.processWriteQueue()
+			go m.processWriteQueue(queue, done)
 		}
 	}()
 
 	for {
 		select {
-		case data := <-m.writeQueue:
+		case data := <-queue:
 			if Port == nil {
 				log.Error("Serial port not connected")
 				continue
@@ -776,7 +805,7 @@ func (m *Module) processWriteQueue() {
 			// Wait a while after sending for the LoRa module to process
 			time.Sleep(50 * time.Millisecond)
 
-		case <-m.writeQueueDone:
+		case <-done:
 			return
 		}
 	}

pkg/config.go

@@ -22,7 +22,9 @@ type Config struct {
 	MQTTUsername         string        `yaml:"mqtt_username"`
 	MQTTPassword         string        `yaml:"mqtt_password" type:"secret"`
 	MQTTTopicPrefix      string        `yaml:"mqtt_topic_prefix"`
-	timeOffAirDefault    time.Duration `yaml:"time_off_air_default"`
+	// WriteResponseTimeout is how long the radio is held idle after each write
+	// transmission waiting for the device's RESPONSE before the next frame goes.
+	WriteResponseTimeout time.Duration `yaml:"write_response_timeout"`
 }
 
 const DefaultDeviceKey = "0301021604050f07e6095a0b0c12630f"
@@ -39,7 +41,7 @@ func (m *Module) DefaultConfig() *Config {
 		MQTTUsername:         "",
 		MQTTPassword:         "",
 		MQTTTopicPrefix:      MQTTTopicPrefix,
-		timeOffAirDefault:    5 * time.Second,
+		WriteResponseTimeout: 5 * time.Second,
 	}
 }
 
@@ -54,6 +56,13 @@ func (m *Module) ValidateAndSetConfig(config []byte) ([]byte, error) {
 	logger.SetLogger(logLevel)
 	newConfig.LogLevel = strings.ToUpper(logLevel.String())
 
+	if newConfig.WriteResponseTimeout <= 0 {
+		newConfig.WriteResponseTimeout = 5 * time.Second
+	}
+	if newConfig.WriteQueueMaxRetries <= 0 {
+		newConfig.WriteQueueMaxRetries = 1
+	}
+
 	keyBytes, err := hex.DecodeString(newConfig.DefaultKey)
 	if err != nil {
 		return nil, err

pkg/dbWriter.go

@@ -1,6 +1,7 @@
 package pkg
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 
@@ -150,6 +151,16 @@ func (m *Module) updateDeviceWrittenPoint(pointIDStr string, value float64, err
 	return nil
 }
 
+// onWriteExhausted is called by the write scheduler once a point's write has
+// used up all its attempts without a device RESPONSE.
+func (m *Module) onWriteExhausted(point *model.Point) {
+	if point.UUID == "" { // synthetic points (e.g. UART ping) are not stored
+		return
+	}
+	err := fmt.Errorf("no response from device after %d write attempts", m.config.WriteQueueMaxRetries)
+	_, _ = m.updateWrittenPointError(point, err)
+}
+
 func selectPointByIoNumber(ioNumber string, device *model.Device) *model.Point {
 	if device == nil {
 		return nil

pkg/enable.go

@@ -22,12 +22,16 @@ func (m *Module) Enable() error {
 	}
 
 	m.initWriteQueue()
+	if m.pointWriteQueueManager != nil {
+		m.pointWriteQueueManager.Stop()
+	}
 	m.pointWriteQueueManager = NewPointWriteQueueManager(
 		m.config.WriteQueueMaxRetries,
-		m.config.timeOffAirDefault,
+		m.config.WriteResponseTimeout,
 		m.getDevice,
 		m.getEncryptionKey,
-		m.WriteToLoRaRaw)
+		m.WriteToLoRaRaw,
+		m.onWriteExhausted)
 
 	if m.config.MQTTEnable && m.mqttClient == nil {
 		m.mqttClient = NewMQTTClient(
@@ -73,12 +77,10 @@ func (m *Module) Disable() error {
 	defer m.mutex.Unlock()
 	log.Info("plugin is disabling...")
 	m.interruptChan <- struct{}{}
-	if m.writeQueue != nil {
-		m.writeQueueDone <- struct{}{}
-		close(m.writeQueueDone)
-		close(m.writeQueue)
-		m.writeQueue = nil
+	if m.pointWriteQueueManager != nil {
+		m.pointWriteQueueManager.Stop()
 	}
+	m.stopWriteQueue()
 
 	if m.mqttClient != nil {
 		m.mqttClient.Disconnect()

pkg/module.go

@@ -17,7 +17,7 @@ type Module struct {
 	pointWriteQueueManager *PointWriteQueueManager
 	writeQueue             chan []byte
 	writeQueueDone         chan struct{}
-	writeQueueInit         sync.Once
+	writeQueueMutex        sync.Mutex
 	mqttClient             *MQTTClient
 }
 

pkg/serial.go

@@ -68,8 +68,16 @@ func (m *Module) SerialClose() error {
 func (m *Module) WriteToLoRaRaw(data []byte) error {
 	m.initWriteQueue() // Make sure the queue is initialized
 
+	if Port == nil {
+		return errors.New("serial port not connected")
+	}
+	queue := m.getWriteQueue()
+	if queue == nil {
+		return errors.New("write queue stopped")
+	}
+
 	select {
-	case m.writeQueue <- data:
+	case queue <- data:
 		return nil
 	case <-time.After(1 * time.Second):
 		return errors.New("write queue full, timeout after 1 second")