Skip to content

File upload: send each batch from the latest ack #18

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
robdobsn merged 1 commit into from
Oct 23, 2026
Merged

File upload: send each batch from the latest ack #18

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 25 additions & 23 deletions src/RaftFileHandler.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,10 @@ async function upload(fileLen: number, block: number, batch: number, lose: numbe
};
}

// Lost blocks (file positions) in a 100-block file of 100-byte blocks, batches of 10
const MID_BATCH = [2300, 5400, 8600];
const BATCH_START = [2000, 5000, 8000];

describe("RaftFileHandler upload", () => {
it("sends each block once when none are lost", async () => {
const r = await upload(5000, 100, 10);
Expand All @@ -139,39 +143,37 @@ describe("RaftFileHandler upload", () => {
expect(r.maxInFlight).toBeLessThanOrEqual(10);
}, 30000);

// Before the change: 88 and 97 blocks sent for 50
// Current firmware (RaftCore from 2026-10-23) acks the first block out of sequence at once and keeps its acks in
// step with the batches: little is sent twice, and a lost block mid-batch needs no timer ack
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[]);
["mid-batch", MID_BATCH, 0],
["at the start of a batch", BATCH_START, 1],
])("recovers from blocks lost %s with current firmware", async (_name, lose, maxTimerAcks) => {
const r = await upload(10000, 100, 10, lose as number[], 15, true);
expect([r.ok, r.intact]).toEqual([true, true]);
expect(r.blocksSent).toBeLessThanOrEqual(r.blocksNeeded + (extra as number));
}, 30000);
expect(r.timerAcks).toBeLessThanOrEqual(maxTimerAcks as number);
expect(r.blocksSent).toBeLessThanOrEqual(r.blocksNeeded + 3 * 10);
expect(r.maxInFlight).toBeLessThanOrEqual(10);
}, 60000);

// 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)
// Older firmware counts the blocks it drops after a loss, so its acks fall part-way through batches. Before this
// handler counted acks arriving while a batch was going out, each later batch then waited for the device's 1s
// timer ack (13 of them after one loss in a 200-block file)
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) => {
["mid-batch", MID_BATCH],
["at the start of a batch", BATCH_START],
])("doesn't wait for timer acks after blocks lost %s with older firmware", 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);
expect(r.maxInFlight).toBeLessThanOrEqual(10);
}, 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);
it.each([[false], [true]])("recovers the last block lost (current firmware %s)", async (currentFirmware) => {
const r = await upload(5000, 100, 10, [4900], 0, currentFirmware);
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);
expect(r.blocksSent).toBe(r.blocksNeeded + 1);
}, 30000);

it("ignores a late ack behind one already received", () => {
const handler = new RaftFileHandler({} as RaftMsgHandler, new RaftCommsStats());
Expand Down
43 changes: 13 additions & 30 deletions src/RaftFileHandler.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,10 +53,6 @@ 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;
Expand All @@ -67,9 +63,6 @@ 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;
Expand Down Expand Up @@ -270,7 +263,6 @@ export default class RaftFileHandler {

this._batchAckReceived = false;
this._ackedFilePos = 0;
this._nextFilePos = 0;

// Send file blocks
let progressUpdateCtr = 0;
Expand All @@ -289,40 +281,35 @@ 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
const batchSize = this._nextFilePos == 0 ? 1 : this._batchAckSize;
const ackedBeforeBatch = this._ackedFilePos;
// Acks that arrive while the batch is going out count
//
// Each batch is sent from the latest ack (go-back-N). Current RaftCore acks the first block out of sequence
// at once and keeps its acks in step with the batches, so after a lost block this resends little and
// doesn't wait for the device's timer ack. (Carrying on past an ack that covers only part of a batch
// resends less with older firmware, but with current firmware it sends blocks the device then drops
// without an ack, and waits 1s for the timer ack after every lost block)
let sendFromPos = this._ackedFilePos;
const batchSize = sendFromPos == 0 ? 1 : this._batchAckSize;
// Acks that arrive while the batch is going out count (the device may ack part-way through a batch)
this._batchAckReceived = false;
this._deviceWaiting = false;
for (
let i = 0;
i < batchSize && this._nextFilePos < fileContents.length;
i < batchSize && sendFromPos < fileContents.length;
i++
) {
// Debug
RaftLog.debug(
`_sendFileContents sendblock pos ${this._nextFilePos} len ${this._fileBlockSize} ackedTo ${this._ackedFilePos} fileLen ${fileContents.length}`,
`_sendFileContents sendblock pos ${sendFromPos} len ${this._fileBlockSize} ackedTo ${this._ackedFilePos} fileLen ${fileContents.length}`,
);
if (!await this._sendFileBlock(fileContents, this._nextFilePos))
if (!await this._sendFileBlock(fileContents, sendFromPos))
return false;
this._nextFilePos = Math.min(fileContents.length, this._nextFilePos + this._fileBlockSize);
sendFromPos += 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
Expand Down Expand Up @@ -399,7 +386,6 @@ export default class RaftFileHandler {
}

// Record
this._lastBlockSentMs = Date.now();
this._commsStats.recordFileBytes(blockEnd - blockStart);

// Add to list of pending messages
Expand All @@ -421,9 +407,6 @@ export default class RaftFileHandler {
// 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}`);
}

Expand Down
Loading