Feature/upload resend after loss

package.json

@@ -1,6 +1,6 @@
 {
     "name": "@robdobsn/raftjs",
-    "version": "2.8.1",
+    "version": "2.9.0",
     "description": "Javascript/TS library for Raft library",
     "main": "dist/web/main.js",
     "types": "dist/web/main.d.ts",

src/RaftFileHandler.test.ts

@@ -0,0 +1,185 @@
+import RaftCommsStats from "./RaftCommsStats";
+import RaftFileHandler from "./RaftFileHandler";
+import RaftMsgHandler from "./RaftMsgHandler";
+import { RaftFileSendType } from "./RaftTypes";
+
+// A device that takes uploads as the firmware's FileUploadOKTOProtocol does: every block it receives counts towards
+// the ack batch, including blocks at the wrong position (which it drops); it acks after the first block, after each
+// batch and after the last block, and on a timer when blocks stop arriving. All its acks look the same
+// ({"req":"ufBlock","rslt":"ok","okto":N}), so the handler only sees the position.
+class SimDevice {
+  data: number[] = [];
+  blocksReceived = 0;
+  timerAcks = 0;
+  private len = 0;
+  private batch = 0;
+  private batchCount = 0;
+  private blockCount = 0;
+  private active = false;
+  private outOfSequenceAcked = false;
+  private timer?: ReturnType<typeof setTimeout>;
+  private lose: Set<number>;
+  handler?: RaftFileHandler;
+
+  // `lose`: file positions whose first block is lost on the way; `timerMs` is the device's (1s).
+  // `ackOutOfSequence`: as RaftCore from 2026-10-23 - the first block out of sequence is acked at once, the rest
+  // aren't counted towards the batch, and a timer ack starts the batch count again
+  constructor(lose: number[], private ackOutOfSequence = false, private ackDelayMs = 2, private timerMs = 1000) {
+    this.lose = new Set(lose);
+  }
+
+  start(len: number, batch: number) {
+    Object.assign(this, { data: [], len, batch, batchCount: 0, blockCount: 0, active: true });
+  }
+
+  block(pos: number, bytes: Uint8Array) {
+    this.blocksReceived++;
+    if (this.lose.has(pos)) {
+      this.lose.delete(pos);
+      return;
+    }
+    if (!this.active) return;
+    if (this.ackOutOfSequence && pos !== this.data.length) {
+      if (!this.outOfSequenceAcked) {
+        this.outOfSequenceAcked = true;
+        this.batchCount = 0;
+        this.ack();
+      }
+      this.armTimer();
+      return;
+    }
+    this.outOfSequenceAcked = false;
+    this.batchCount++;
+    let isFinal = false;
+    if (pos === this.data.length) {
+      this.data.push(...bytes);
+      this.blockCount++;
+      isFinal = this.data.length >= this.len;
+    }
+    if (this.batchCount === this.batch || this.blockCount === 1 || isFinal) {
+      this.batchCount = 0;
+      this.ack();
+    }
+    if (isFinal) {
+      this.active = false;
+      clearTimeout(this.timer);
+    } else {
+      this.armTimer();
+    }
+  }
+
+  private ack() {
+    const okto = this.data.length;
+    setTimeout(() => this.handler?.onOktoMsg(okto), this.ackDelayMs);
+  }
+
+  private armTimer() {
+    clearTimeout(this.timer);
+    this.timer = setTimeout(() => {
+      if (!this.active) return;
+      this.timerAcks++;
+      if (this.ackOutOfSequence) this.batchCount = 0;
+      this.ack();
+      this.armTimer();
+    }, this.timerMs);
+  }
+
+  stop() {
+    this.active = false;
+    clearTimeout(this.timer);
+  }
+}
+
+// Upload a file to a SimDevice. `sendMs` is how long each block takes to send (0: an instant link)
+async function upload(fileLen: number, block: number, batch: number, lose: number[] = [], sendMs = 0,
+                      ackOutOfSequence = false) {
+  const device = new SimDevice(lose, ackOutOfSequence);
+  const msgHandler = {
+    sendRICREST: async (cmd: string) => {
+      const json = JSON.parse(cmd);
+      if (json.cmdName === "ufStart") {
+        device.start(json.fileLen, batch);
+        return { rslt: "ok", batchMsgSize: block, batchAckSize: batch, streamID: 1 };
+      }
+      return { rslt: "ok" };
+    },
+    sendFileBlock: async (bytes: Uint8Array, pos: number) => {
+      // Blocks in flight: sent beyond the position the handler knows is acked
+      const acked = (handler as unknown as { _ackedFilePos: number })._ackedFilePos;
+      maxInFlight = Math.max(maxInFlight, Math.ceil((pos + bytes.length - acked) / block));
+      if (sendMs > 0) await new Promise((resolve) => setTimeout(resolve, sendMs));
+      device.block(pos, bytes);
+      return true;
+    },
+  } as unknown as RaftMsgHandler;
+  let maxInFlight = 0;
+  const handler = new RaftFileHandler(msgHandler, new RaftCommsStats());
+  device.handler = handler;
+  const contents = Uint8Array.from({ length: fileLen }, (_, i) => (i * 7) & 0xff);
+  const start = Date.now();
+  const ok = await handler.fileSend("test.bin", RaftFileSendType.NORMAL_FILE, "fs", contents, undefined);
+  device.stop();
+  return {
+    ok,
+    intact: device.data.length === fileLen && device.data.every((b, i) => b === contents[i]),
+    blocksSent: device.blocksReceived,
+    blocksNeeded: Math.ceil(fileLen / block),
+    timerAcks: device.timerAcks,
+    maxInFlight,
+    ms: Date.now() - start,
+  };
+}
+
+describe("RaftFileHandler upload", () => {
+  it("sends each block once when none are lost", async () => {
+    const r = await upload(5000, 100, 10);
+    expect([r.ok, r.intact]).toEqual([true, true]);
+    expect(r.blocksSent).toBe(r.blocksNeeded);
+    expect(r.timerAcks).toBe(0);
+    expect(r.maxInFlight).toBeLessThanOrEqual(10);
+  }, 30000);
+
+  // Before the change: 88 and 97 blocks sent for 50
+  it.each([
+    ["one lost block", [1300], 8],
+    ["two lost blocks", [1300, 3700], 21],
+    ["the last block lost", [4900], 1],
+  ])("resends little after %s", async (_name, lose, extra) => {
+    const r = await upload(5000, 100, 10, lose as number[]);
+    expect([r.ok, r.intact]).toEqual([true, true]);
+    expect(r.blocksSent).toBeLessThanOrEqual(r.blocksNeeded + (extra as number));
+  }, 30000);
+
+  // Before the change, on a link where each block takes time to send, an ack arriving while a batch was going out
+  // was ignored, so after one loss every later batch waited for the device's 1s timer ack (13 of them here)
+  it.each([
+    ["one lost block", [3000]],
+    ["three lost blocks", [2000, 5000, 8000]],
+  ])("doesn't wait for timer acks after %s on a slow link", async (_name, lose) => {
+    const r = await upload(10000, 100, 10, lose as number[], 15);
+    expect([r.ok, r.intact]).toEqual([true, true]);
+    expect(r.timerAcks).toBeLessThanOrEqual(1);
+    // At most two batches beyond the last ack received
+    expect(r.maxInFlight).toBeLessThanOrEqual(20);
+  }, 60000);
+
+  // Firmware that acks the first block out of sequence at once (RaftCore from 2026-10-23): acks stay in step with
+  // the batches, so no more than one batch is in flight and few blocks are sent twice
+  it("works with firmware that acks a block out of sequence at once", async () => {
+    const r = await upload(10000, 100, 10, [2000, 5000, 8000], 15, true);
+    expect([r.ok, r.intact]).toEqual([true, true]);
+    expect(r.timerAcks).toBeLessThanOrEqual(1);
+    expect(r.maxInFlight).toBeLessThanOrEqual(10);
+    expect(r.blocksSent).toBeLessThanOrEqual(r.blocksNeeded + 3 * 10);
+  }, 60000);
+
+  it("ignores a late ack behind one already received", () => {
+    const handler = new RaftFileHandler({} as RaftMsgHandler, new RaftCommsStats());
+    const state = handler as unknown as { _ackedFilePos: number; _batchAckReceived: boolean };
+    handler.onOktoMsg(800);
+    state._batchAckReceived = false;
+    handler.onOktoMsg(400);
+    expect(state._ackedFilePos).toBe(800);
+    expect(state._batchAckReceived).toBe(false);
+  });
+});

src/RaftFileHandler.ts

@@ -53,6 +53,10 @@ export default class RaftFileHandler {
 
   // Timeouts
   private BLOCK_ACK_TIMEOUT_MS = 30000;
+  // The device acks on a timer when it has received no block for 1s (5s before the second block). Its timer acks look
+  // like any other, so an ack this long after the last block was sent is taken as one: the device is waiting for
+  // the position it acks
+  private DEVICE_WAITING_ACK_MS = 700;
 
   // Contents of file to send
   private _requestedFileBlockSize = 500;
@@ -63,6 +67,9 @@ export default class RaftFileHandler {
   // File sending flow control
   private _sendWithoutBatchAcks = false;
   private _ackedFilePos = 0;
+  private _nextFilePos = 0;
+  private _lastBlockSentMs = 0;
+  private _deviceWaiting = false;
   private _batchAckReceived = false;
   private _isTxCancelled = false;
   private _fileTxStreamID = 0;
@@ -263,6 +270,7 @@ export default class RaftFileHandler {
 
     this._batchAckReceived = false;
     this._ackedFilePos = 0;
+    this._nextFilePos = 0;
 
     // Send file blocks
     let progressUpdateCtr = 0;
@@ -281,31 +289,40 @@ export default class RaftFileHandler {
         // NOTE: first batch MUST be of size 1 (not _batchAckSize) because Raft performs a long-running
         // blocking task immediately after receiving the first message in a firmware
         // update - although this could be relaxed for non-firmware update file uploads
-        let sendFromPos = this._ackedFilePos;
-        const batchSize = sendFromPos == 0 ? 1 : this._batchAckSize;
+        const batchSize = this._nextFilePos == 0 ? 1 : this._batchAckSize;
+        const ackedBeforeBatch = this._ackedFilePos;
+        // Acks that arrive while the batch is going out count
+        this._batchAckReceived = false;
+        this._deviceWaiting = false;
         for (
           let i = 0;
-          i < batchSize && sendFromPos < fileContents.length;
+          i < batchSize && this._nextFilePos < fileContents.length;
           i++
         ) {
-          // Clear old batch acks
-          if (i == batchSize - 1) {
-            this._batchAckReceived = false;
-          }
           // Debug
            RaftLog.debug(
-             `_sendFileContents sendblock pos ${sendFromPos} len ${this._fileBlockSize} ackedTo ${this._ackedFilePos} fileLen ${fileContents.length}`,
+             `_sendFileContents sendblock pos ${this._nextFilePos} len ${this._fileBlockSize} ackedTo ${this._ackedFilePos} fileLen ${fileContents.length}`,
            );
-          if (!await this._sendFileBlock(fileContents, sendFromPos))
+          if (!await this._sendFileBlock(fileContents, this._nextFilePos))
             return false;
-          sendFromPos += this._fileBlockSize;
+          this._nextFilePos = Math.min(fileContents.length, this._nextFilePos + this._fileBlockSize);
         }
 
         // Wait for response (there is a timeout at the ESP end to ensure a response is always returned
         // even if blocks are dropped on reception at ESP) - the timeout here is for these responses
         // being dropped
         await this.batchAck(this.BLOCK_ACK_TIMEOUT_MS);
         progressUpdateCtr += this._batchAckSize;
+
+        // Where to send from next. An ack with progress can be for part of what was sent: the device acks every
+        // batchAckSize blocks it receives, counting blocks it dropped, so after a lost block its acks fall in the
+        // middle of batches - carry on from what was sent, as going back would resend blocks the device has. An
+        // ack without progress, or a timer ack (the device is waiting), means blocks were lost: go back to it
+        if (this._ackedFilePos > ackedBeforeBatch && !this._deviceWaiting) {
+          this._nextFilePos = Math.max(this._nextFilePos, this._ackedFilePos);
+        } else {
+          this._nextFilePos = this._ackedFilePos;
+        }
       }
 
       // Show progress
@@ -382,6 +399,7 @@ export default class RaftFileHandler {
     }
 
     // Record
+    this._lastBlockSentMs = Date.now();
     this._commsStats.recordFileBytes(blockEnd - blockStart);
 
     // Add to list of pending messages
@@ -395,9 +413,17 @@ export default class RaftFileHandler {
   }
 
   onOktoMsg(fileOkTo: number) {
+    // A late ack behind one already received says nothing new
+    if (fileOkTo < this._ackedFilePos) {
+      RaftLog.verbose(`onOktoMsg ignored late OkTo ${fileOkTo} acked ${this._ackedFilePos}`);
+      return;
+    }
     // Get how far we've progressed in file
     this._ackedFilePos = fileOkTo;
     this._batchAckReceived = true;
+    if (Date.now() - this._lastBlockSentMs >= this.DEVICE_WAITING_ACK_MS) {
+      this._deviceWaiting = true;
+    }
     RaftLog.verbose(`onOktoMsg received file up to ${this._ackedFilePos}`);
   }