fix: make radio TX queue flush-safe and report job outcomes

- Serialize queueTx and flushTx so a queued job can't be lost to a flush.
- Settings changes preempt only the job in flight; the queue survives.
- Flush drains the queue and reports each job as dropped.
- Track suspend reasons (scan, OTA) separately.
- Report started/dropped for every job on a single event queue.
- Add priority jobs that go to the front and may use reserved slots.
- Remove the MAX_TX_REPEATS clamp.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/Somfy.cpp

@@ -46,23 +46,48 @@ uint8_t rxmode = 0;  // Indicates whether the radio is in receive mode.  Just to
 // Serializes all access to the CC1101 (SPI) and the TX/RX pins between the
 // radio task, the main loop and the async web server task.
 static SemaphoreHandle_t radioMutex = nullptr;
+// Makes queueing a job and flushing the queue atomic with respect to each other
+// so a job queueTx() reported as queued can't be silently lost to a flush.
+static SemaphoreHandle_t txQueueMutex = nullptr;
 static QueueHandle_t txJobQueue = nullptr;
-static QueueHandle_t txStartQueue = nullptr;
+static QueueHandle_t txEventQueue = nullptr;
 static TaskHandle_t txTaskHandle = nullptr;
-// Bumped by flushTx().  Queued jobs from an older generation are dropped and a
+// Bumped by flushTx().  A job from an older generation that the radio task
-// job in flight stops after its current frame so nobody waits on the radio lock
+// already took off the queue is dropped, and one in flight stops after its
-// for a whole transmission.
+// current frame.
 static std::atomic<uint32_t> txGen{0};
-// Set while the radio must not transmit at all (frequency scan, OTA).
+// Number of tasks waiting for the radio to reconfigure it.  A job in flight
-static volatile bool txSuspended = false;
+// stops after its current frame so they never wait for a whole transmission.
+static std::atomic<uint32_t> txPreemptWaiters{0};
+// TX_SUSPEND_* reasons the radio must not transmit for (frequency scan, OTA).
+static std::atomic<uint32_t> txSuspendMask{0};
 struct RadioLock {
   bool locked = true;  // Before the mutex exists there is nothing to serialize against.
   bool owned = false;
-  RadioLock(TickType_t wait = portMAX_DELAY) {
+  // preempt cuts short a transmission in flight instead of waiting for it.
-    if(radioMutex) locked = owned = xSemaphoreTakeRecursive(radioMutex, wait) == pdTRUE;
+  RadioLock(TickType_t wait = portMAX_DELAY, bool preempt = false) {
+    if(!radioMutex) return;
+    if(preempt) ++txPreemptWaiters;
+    locked = owned = xSemaphoreTakeRecursive(radioMutex, wait) == pdTRUE;
+    if(preempt) --txPreemptWaiters;
   }
   ~RadioLock() { if(owned) xSemaphoreGiveRecursive(radioMutex); }
 };
+struct TxQueueLock {
+  TxQueueLock() { if(txQueueMutex) xSemaphoreTake(txQueueMutex, portMAX_DELAY); }
+  ~TxQueueLock() { if(txQueueMutex) xSemaphoreGive(txQueueMutex); }
+};
+static void postTxEvent(tx_event_t type, const somfy_tx_job_t &job, uint32_t at) {
+  // Repeater frames don't belong to any of our remotes.
+  if(job.raw || !txEventQueue) return;
+  somfy_tx_event_t evt;
+  evt.type = type;
+  evt.cmd = job.frame.cmd;
+  evt.remoteAddress = job.frame.remoteAddress;
+  evt.queuedAt = job.queuedAt;
+  evt.at = at;
+  if(xQueueSend(txEventQueue, &evt, 0) != pdTRUE) ESP_LOGW(TAG, "Radio TX event queue full");
+}
 // Encodes the first frame of a transmission into the job.  Encoding here keeps
 // its side effects on the frame (valid flag, default step size) visible to the
 // caller and the radio task only re-encodes the 80-bit repeat frames.
@@ -3921,7 +3946,6 @@ void SomfyRemote::repeatFrame(uint8_t repeat) {
     this->triggerGPIOs(this->lastFrame);
     return;
   }
-  repeat = txRepeats(repeat);
   // The first frame is encoded with the current repeat counter and the radio
   // task counts up from there for each repeat frame.
   somfy_tx_job_t job;
@@ -3931,10 +3955,11 @@ void SomfyRemote::repeatFrame(uint8_t repeat) {
   somfy.transceiver.queueTx(job);
   //somfy.processFrame(this->lastFrame, true);
 }
-bool SomfyShadeController::sendFrame(somfy_frame_t &frame, uint8_t repeat, uint16_t delayMs) {
+bool SomfyShadeController::sendFrame(somfy_frame_t &frame, uint8_t repeat, uint16_t delayMs, bool priority) {
   somfy_tx_job_t job;
-  encodeTxJob(job, frame, txRepeats(repeat));
+  encodeTxJob(job, frame, repeat);
   job.firstRepeat = 1;
+  job.priority = priority;
   if(delayMs > 0) job.notBefore = millis() + delayMs;
   return this->transceiver.queueTx(job);
 }
@@ -4421,8 +4446,8 @@ uint32_t lastScan = 0;
 void Transceiver::beginFrequencyScan() {
   // The scan retunes the radio so nothing may transmit until it ends.  This
   // also aborts a transmission in flight so the lock below is never held long.
-  if(this->config.enabled) this->suspendTx();
+  if(this->config.enabled) this->suspendTx(TX_SUSPEND_SCAN);
-  RadioLock lock;
+  RadioLock lock(portMAX_DELAY, true);
   if(this->config.enabled) {
     this->disableReceive();
     rxmode = 3;
@@ -4473,7 +4498,7 @@ void Transceiver::endFrequencyScan() {
     if(interruptPin > 0) detachInterrupt(interruptPin); 
     interruptPin = 0;
     this->config.apply();
-    this->resumeTx();
+    this->resumeTx(TX_SUSPEND_SCAN);
     this->emitFrequencyScan();
   }
 }
@@ -4606,11 +4631,11 @@ bool Transceiver::fromJSON(JsonObject& obj) {
     if (!obj["config"].isNull()) {
       JsonObject objConfig = obj["config"];
       // Parse into a copy so the radio task never sees a half updated config,
-      // then swap it in once any transmission in flight has stopped.
+      // then swap it in once any transmission in flight has stopped.  Queued
+      // jobs are kept and go out with the new settings.
       transceiver_config_t cfg = this->config;
       cfg.fromJSON(objConfig);
-      this->flushTx();
+      RadioLock lock(portMAX_DELAY, true);
-      RadioLock lock;
       if(this->config.enabled && this->config.TXPin != cfg.TXPin) pinMode(this->config.TXPin, INPUT);
       this->config = cfg;
     }
@@ -4636,8 +4661,8 @@ bool Transceiver::save() {
 bool Transceiver::end() {
     // Nothing may transmit (and re-enable receive) after this.  It is not
     // resumed since the only caller is OTA which always reboots.
-    this->suspendTx();
+    this->suspendTx(TX_SUSPEND_OTA);
-    RadioLock lock;
+    RadioLock lock(portMAX_DELAY, true);
     this->disableReceive();
     return true;
 }
@@ -4872,10 +4897,10 @@ void transceiver_config_t::load() {
     //this->printBuffer = somfy.transceiver.printBuffer;
 }
 void transceiver_config_t::apply() {
-    // Frames queued under the old settings are discarded and any frame that is
+    // A transmission in flight stops after its current frame so the radio can
-    // in flight completes before the radio is reconfigured.
+    // be reconfigured.  Queued frames are already encoded and go out with the
-    somfy.transceiver.flushTx();
+    // new settings.
-    RadioLock lock;
+    RadioLock lock(portMAX_DELAY, true);
     somfy.transceiver.disableReceive();
     bit_length = this->type;    
     if(this->enabled) {
@@ -4977,6 +5002,7 @@ void transceiver_config_t::apply() {
 }
 bool Transceiver::begin() {
     if(!radioMutex) radioMutex = xSemaphoreCreateRecursiveMutex();
+    if(!txQueueMutex) txQueueMutex = xSemaphoreCreateMutex();
     this->config.load();
     this->config.apply();
     rx_queue.init();
@@ -4985,11 +5011,16 @@ bool Transceiver::begin() {
 }
 void Transceiver::loop() {
   somfy_rx_t rx;
-  // Jobs that went on air late.  The shades were set moving when the job was
+  somfy_tx_event_t evt;
-  // queued so line their movement up with the motor.
+  while(txEventQueue && xQueueReceive(txEventQueue, &evt, 0) == pdTRUE) {
-  somfy_tx_start_t started;
+    if(evt.type == tx_event_t::started) {
-  while(txStartQueue && xQueueReceive(txStartQueue, &started, 0) == pdTRUE)
+      // The shades were set moving when the job was queued so line their
-    somfy.shiftMovementStart(started.remoteAddress, started.queuedAt, started.startedAt);
+      // movement up with the motor.
+      somfy.shiftMovementStart(evt.remoteAddress, evt.queuedAt, evt.at);
+    }
+    else
+      ESP_LOGW(TAG, "Radio TX dropped %s for %d", translateSomfyCommand(evt.cmd).c_str(), evt.remoteAddress);
+  }
   if (noiseDetected && rxmode != 3 && this->config.noiseDetection) {
     if (millis() - noiseStart > 100) {
       gpio_intr_enable((gpio_num_t)interruptPin);
@@ -5042,30 +5073,44 @@ bool Transceiver::queueTx(somfy_tx_job_t &job) {
     ESP_LOGE(TAG, "Radio task not started, dropping transmission");
     return false;
   }
-  if(txSuspended) {
+  TxQueueLock lock;
+  if(txSuspendMask) {
     ESP_LOGW(TAG, "Radio TX suspended, dropping transmission");
     return false;
   }
+  // Normal jobs leave the reserved slots free for stops.
+  if(uxQueueSpacesAvailable(txJobQueue) <= (job.priority ? 0 : TX_RESERVED_SLOTS)) {
+    ESP_LOGE(TAG, "Radio TX queue full, dropping transmission");
+    return false;
+  }
   job.gen = txGen;
   job.queuedAt = millis();
-  if(xQueueSend(txJobQueue, &job, 0) != pdTRUE) {
+  // We are the only producer while holding the lock so the space check holds.
-    ESP_LOGE(TAG, "Radio TX queue full, dropping transmission");
+  if(job.priority) xQueueSendToFront(txJobQueue, &job, 0);
-    return false;
+  else xQueueSendToBack(txJobQueue, &job, 0);
-  }
   return true;
 }
-bool Transceiver::canQueueTx() {
+bool Transceiver::canQueueTx(bool priority) {
-  return txJobQueue && txTaskHandle && !txSuspended && uxQueueSpacesAvailable(txJobQueue) > 0;
+  return txJobQueue && txTaskHandle && !txSuspendMask && uxQueueSpacesAvailable(txJobQueue) > (priority ? 0 : TX_RESERVED_SLOTS);
 }
 void Transceiver::flushTx() {
+  TxQueueLock lock;
   ++txGen;
-  if(txJobQueue) xQueueReset(txJobQueue);
+  if(!txJobQueue) return;
+  // Drain rather than reset so every discarded job is reported as dropped.
+  somfy_tx_job_t job;
+  const uint32_t now = millis();
+  while(xQueueReceive(txJobQueue, &job, 0) == pdTRUE) postTxEvent(tx_event_t::dropped, job, now);
 }
-void Transceiver::suspendTx() {
+void Transceiver::suspendTx(uint8_t reason) {
-  txSuspended = true;
+  {
+    // Set under the queue lock so nothing is queued between this and the flush.
+    TxQueueLock lock;
+    txSuspendMask |= reason;
+  }
   this->flushTx();
 }
-void Transceiver::resumeTx() { txSuspended = false; }
+void Transceiver::resumeTx(uint8_t reason) { txSuspendMask &= ~(uint32_t)reason; }
 // cpt_synchro_hw is updated by the RX interrupt and read here from the radio task.
 static bool rxInProgress() { return *(volatile uint8_t *)&somfy_rx.cpt_synchro_hw != 0; }
 void Transceiver::transmitJob(somfy_tx_job_t &job) {
@@ -5073,8 +5118,9 @@ void Transceiver::transmitJob(somfy_tx_job_t &job) {
   this->beginTransmit();
   this->sendFrame(tx.payload, tx.hwsync, tx.bit_length);
   for(uint8_t i = 0; i < job.repeats; i++) {
-    // Stop early when flushed so whoever flushed gets the radio within a frame.
+    // Stop early when flushed or when someone is waiting to reconfigure the
-    if(job.gen != txGen) break;
+    // radio so they get it within a frame.
+    if(job.gen != txGen || txPreemptWaiters > 0) break;
     // For each 80-bit frame we need to adjust the byte encoding for the
     // silence.
     if(tx.bit_length == 80) job.frame.encode80BitFrame(tx.payload, (uint8_t)(job.firstRepeat + i));
@@ -5087,7 +5133,11 @@ void Transceiver::txTask(void *arg) {
   somfy_tx_job_t job;
   for(;;) {
     if(xQueueReceive(txJobQueue, &job, portMAX_DELAY) != pdTRUE) continue;
-    if(txSuspended || job.gen != txGen) continue;
+    // Flushed after we took it off the queue.
+    if(txSuspendMask || job.gen != txGen) {
+      postTxEvent(tx_event_t::dropped, job, millis());
+      continue;
+    }
     if(job.notBefore != 0) {
       int32_t wait = (int32_t)(job.notBefore - millis());
       if(wait > 0) vTaskDelay(pdMS_TO_TICKS(wait));
@@ -5100,15 +5150,11 @@ void Transceiver::txTask(void *arg) {
     {
       RadioLock lock;
       // The queue may have been flushed while we waited.
-      if(txSuspended || job.gen != txGen) continue;
+      if(txSuspendMask || job.gen != txGen) {
-      const uint32_t startedAt = millis();
+        postTxEvent(tx_event_t::dropped, job, millis());
-      if(!job.raw && txStartQueue && startedAt - job.queuedAt > TX_QUEUE_DELAY * 2) {
+        continue;
-        somfy_tx_start_t started;
-        started.remoteAddress = job.frame.remoteAddress;
-        started.queuedAt = job.queuedAt;
-        started.startedAt = startedAt;
-        xQueueSend(txStartQueue, &started, 0);
       }
+      postTxEvent(tx_event_t::started, job, millis());
       trans->transmitJob(job);
     }
     // Give the motors a full frame beat between separate transmissions.
@@ -5118,7 +5164,8 @@ void Transceiver::txTask(void *arg) {
 void Transceiver::startTxTask() {
   if(txTaskHandle) return;
   if(!txJobQueue) txJobQueue = xQueueCreate(TX_JOB_QUEUE_SIZE, sizeof(somfy_tx_job_t));
-  if(!txStartQueue) txStartQueue = xQueueCreate(TX_JOB_QUEUE_SIZE, sizeof(somfy_tx_start_t));
+  // Every queued job reports once, plus room for the jobs a flush drains.
+  if(!txEventQueue) txEventQueue = xQueueCreate(TX_JOB_QUEUE_SIZE * 2, sizeof(somfy_tx_event_t));
   if(!txJobQueue) {
     ESP_LOGE(TAG, "Unable to create radio TX queue");
     return;

src/Somfy.h

@@ -201,11 +201,12 @@ struct somfy_frame_t {
     bool isSynonym(somfy_frame_t &f);
     void copy(somfy_frame_t &f);
 };
-// Upper bound on repeats for a single transmission.  Each frame is ~140ms of
-// radio time so this keeps a single job to roughly 6 seconds.
-#define MAX_TX_REPEATS 40
 #define TX_JOB_QUEUE_SIZE 32
-inline uint8_t txRepeats(uint8_t repeat) { return repeat > MAX_TX_REPEATS ? MAX_TX_REPEATS : repeat; }
+// Slots only priority jobs (stops) may use so a stop always fits in the queue.
+#define TX_RESERVED_SLOTS 4
+// Reasons the radio must not transmit.  TX resumes once every reason is cleared.
+#define TX_SUSPEND_SCAN 0x01
+#define TX_SUSPEND_OTA  0x02
 // A transmission handed off to the radio task.  The first frame is encoded by
 // the producer into tx.  The frame is a copy so the radio task never touches the
 // remote's lastFrame and is only used to re-encode 80-bit repeat frames.
@@ -215,15 +216,22 @@ struct somfy_tx_job_t {
   uint8_t repeats = 0;      // Number of repeat frames after the first frame.
   uint8_t firstRepeat = 1;  // 80-bit repeat counter used for the first repeat frame.
   bool raw = false;         // Repeater frame.  Waits for the receiver to be idle.
+  bool priority = false;    // Goes to the front of the queue and may use the reserved slots.
   uint32_t gen = 0;         // Flush generation when queued.  Stale jobs are dropped.
   uint32_t queuedAt = 0;    // millis() when queued.
   uint32_t notBefore = 0;   // millis() before which the job must not go on air (0 = none).
 };
-// Reported by the radio task when a queued job finally goes on air.
+enum class tx_event_t : uint8_t {
-struct somfy_tx_start_t {
+  started,  // The job is going on air.
+  dropped   // The job was flushed and will never go on air.
+};
+// Reported by the radio task for every queued non-repeater job.
+struct somfy_tx_event_t {
+  tx_event_t type = tx_event_t::started;
+  somfy_commands cmd = somfy_commands::My;
   uint32_t remoteAddress = 0;
   uint32_t queuedAt = 0;
-  uint32_t startedAt = 0;
+  uint32_t at = 0;          // millis() when the job started or was dropped.
 };
 
 class SomfyRoom {
@@ -536,10 +544,10 @@ class Transceiver {
     somfy_frame_t& lastFrame();
     void sendFrame(byte *frame, uint8_t sync, uint8_t bitLength = 56);
     bool queueTx(somfy_tx_job_t &job);
-    bool canQueueTx();
+    bool canQueueTx(bool priority = false);
     void flushTx();
-    void suspendTx();
+    void suspendTx(uint8_t reason);
-    void resumeTx();
+    void resumeTx(uint8_t reason);
     void beginTransmit();
     void endTransmit();
     void emitFrame(somfy_frame_t *frame, somfy_rx_t *rx = nullptr);
@@ -595,7 +603,7 @@ class SomfyShadeController {
     SomfyGroup * getGroupById(uint8_t groupId);
     SomfyShade * findShadeByRemoteAddress(uint32_t address);
     SomfyGroup * findGroupByRemoteAddress(uint32_t address);
-    bool sendFrame(somfy_frame_t &frame, uint8_t repeats = 0, uint16_t delayMs = 0);
+    bool sendFrame(somfy_frame_t &frame, uint8_t repeats = 0, uint16_t delayMs = 0, bool priority = false);
     void processFrame(somfy_frame_t &frame, bool internal = false);
     void shiftMovementStart(uint32_t remoteAddress, uint32_t queuedAt, uint32_t startedAt);
     void emitState(uint8_t num = 255);