From 70cf04c3ff000000aadc2038bbf7c6aca45ac728 Mon Sep 17 00:00:00 2001
From: arif-dewi <11630380+arif-dewi@users.noreply.github.com>
Date: Thu, 27 Aug 2026 12:00:49 +0400
Subject: [PATCH 1/2] fix(miner-logs): declare the real payload format on the
 log file leg
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The file leg hardcoded `.log` and `application/octet-stream` for every
download. The bytes are whatever the miner produced: some models stream a
plain-text log, Whatsminers stream a gzipped tar of their log directory
(<ip>.logs/board_stat0..3.log, power.log, miner-state.log, …). Clients that
trusted the declared name saved an archive as .log and opened it as garbage;
the frontend had already been flipped between .tar.gz and .log once each way
because the constant can only ever be right for one fleet.

The action result carries no format field (coreKey, byteLength, expiresAt,
minerId), so read the leading bytes instead:

- lib/payloadFormat.js detects gzip/zip/text and, for gzip, inflates only the
  first tar header to tell a gzipped tar from a bare .gz — never claiming a
  format it has not seen. Content-Type now follows the same detection.
- The peeked chunk is re-emitted ahead of the rest via prependChunk(), which
  honours consumer backpressure. stream.unshift() cannot be used here: once
  the peek has consumed the source's last chunk it silently drops the head,
  truncating small logs.
- A source failure before the first chunk returns 500 with a JSON body rather
  than a truncated stream under a confident content-type.

Content-Length, Cache-Control and the ownership/expiry checks are unchanged,
and the body is byte-identical to what the miner sent.

Note: tests/unit/handlers/finance.handlers.test.js and the two admin_external
cases in auth/users routes already fail on develop, unrelated to this change.
---
 .../unit/handlers/minerLogs.handlers.test.js  | 107 +++++++++++
 tests/unit/lib/payloadFormat.test.js          | 172 ++++++++++++++++++
 .../lib/server/handlers/actions.handlers.js   |  25 ++-
 workers/lib/server/lib/payloadFormat.js       | 139 ++++++++++++++
 4 files changed, 438 insertions(+), 5 deletions(-)
 create mode 100644 tests/unit/lib/payloadFormat.test.js
 create mode 100644 workers/lib/server/lib/payloadFormat.js

diff --git a/tests/unit/handlers/minerLogs.handlers.test.js b/tests/unit/handlers/minerLogs.handlers.test.js
index d165595..a8879a5 100644
--- a/tests/unit/handlers/minerLogs.handlers.test.js
+++ b/tests/unit/handlers/minerLogs.handlers.test.js
@@ -1,5 +1,7 @@
 'use strict'
 
+const zlib = require('zlib')
+const { Readable } = require('streamx')
 const test = require('brittle')
 const {
   startMinerLogDownload,
@@ -14,13 +16,19 @@ const {
 function makeMockReply () {
   let _code = 200
   let _body = null
+  const _headers = {}
   const reply = {
     get statusCode () { return _code },
     get body () { return _body },
+    get headers () { return _headers },
     code (statusCode) {
       _code = statusCode
       return reply
     },
+    header (name, value) {
+      _headers[name.toLowerCase()] = value
+      return reply
+    },
     send (body) {
       _body = body
       return body
@@ -476,6 +484,105 @@ test('getMinerLogFile - returns 503 when the log peer is unreachable', async (t)
   t.pass()
 })
 
+// The miner decides the payload format and the action result does not say which, so the file
+// leg reads the leading bytes and declares what it actually found. See lib/payloadFormat.
+
+function makeLogStream (payload) {
+  let sent = false
+  return new Readable({
+    read (cb) {
+      if (!sent) {
+        sent = true
+        this.push(payload)
+      } else {
+        this.push(null)
+      }
+      cb(null)
+    }
+  })
+}
+
+function makeFileLegCtx (payload) {
+  return {
+    dataProxy: {
+      requestData: async () => [makeActionResult()]
+    },
+    logDownloader: {
+      stream: async () => makeLogStream(payload)
+    }
+  }
+}
+
+async function drain (stream) {
+  const chunks = []
+  for await (const chunk of stream) chunks.push(chunk)
+  return Buffer.concat(chunks)
+}
+
+test('getMinerLogFile - declares .tar.gz for a gzipped tar payload', async (t) => {
+  const tar = Buffer.alloc(1024)
+  tar.write('10.0.0.1.logs/', 0, 'latin1')
+  tar.write('ustar', 257, 'latin1')
+  const payload = zlib.gzipSync(tar)
+
+  const reply = makeMockReply()
+  await getMinerLogFile(makeFileLegCtx(payload), makeMockReq('miner-001', '42'), reply)
+
+  t.is(
+    reply.headers['content-disposition'],
+    'attachment; filename="miner-log-miner-001-42.tar.gz"',
+    'should name the archive .tar.gz'
+  )
+  t.is(reply.headers['content-type'], 'application/gzip', 'should declare gzip')
+  t.alike(await drain(reply.body), payload, 'should stream every byte, peek included')
+})
+
+test('getMinerLogFile - declares .log for a plain-text payload', async (t) => {
+  const payload = Buffer.from('[board0]\npass = 1\n')
+
+  const reply = makeMockReply()
+  await getMinerLogFile(makeFileLegCtx(payload), makeMockReq('miner-001', '42'), reply)
+
+  t.is(
+    reply.headers['content-disposition'],
+    'attachment; filename="miner-log-miner-001-42.log"',
+    'should name a text log .log'
+  )
+  t.is(reply.headers['content-type'], 'text/plain; charset=utf-8', 'should declare text')
+  t.alike(await drain(reply.body), payload, 'should stream every byte, peek included')
+})
+
+test('getMinerLogFile - keeps the byte length and no-store headers', async (t) => {
+  const reply = makeMockReply()
+  await getMinerLogFile(
+    makeFileLegCtx(Buffer.from('log line')),
+    makeMockReq('miner-001', '42'),
+    reply
+  )
+
+  t.is(reply.headers['content-length'], 1024, 'should send the length from the action meta')
+  t.is(reply.headers['cache-control'], 'no-store', 'should keep the log out of caches')
+})
+
+test('getMinerLogFile - returns 500 when the stream errors before the first byte', async (t) => {
+  const ctx = {
+    dataProxy: {
+      requestData: async () => [makeActionResult()]
+    },
+    logDownloader: {
+      stream: async () => new Readable({
+        read (cb) { cb(new Error('ERR_LOG_PEER_TIMEOUT')) }
+      })
+    }
+  }
+
+  const reply = makeMockReply()
+  await getMinerLogFile(ctx, makeMockReq('miner-001', '42'), reply)
+
+  t.is(reply.statusCode, 500, 'should return 500')
+  t.is(reply.body.error, 'ERR_LOG_PEER_TIMEOUT', 'should propagate the stream error as JSON')
+})
+
 test('getMinerLogDownloadStatus - finds successful result across multiple racks', async (t) => {
   const expiresAt = Date.now() + 3600000
   const action = {
diff --git a/tests/unit/lib/payloadFormat.test.js b/tests/unit/lib/payloadFormat.test.js
new file mode 100644
index 0000000..a951fcd
--- /dev/null
+++ b/tests/unit/lib/payloadFormat.test.js
@@ -0,0 +1,172 @@
+'use strict'
+
+const zlib = require('zlib')
+const { Readable } = require('streamx')
+const test = require('brittle')
+
+const {
+  detectPayloadFormat,
+  peekFirstChunk,
+  prependChunk
+} = require('../../../workers/lib/server/lib/payloadFormat')
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Helpers
+// ─────────────────────────────────────────────────────────────────────────────
+
+// A 512-byte POSIX tar header carries "ustar" at offset 257, followed by payload blocks.
+function tarBytes (name = '10.0.0.1.logs/') {
+  const tar = Buffer.alloc(1024)
+  tar.write(name, 0, 'latin1')
+  tar.write('ustar', 257, 'latin1')
+  return tar
+}
+
+function makeStream (chunks) {
+  let index = 0
+  return new Readable({
+    read (cb) {
+      if (index < chunks.length) this.push(Buffer.from(chunks[index++]))
+      else this.push(null)
+      cb(null)
+    }
+  })
+}
+
+async function drain (stream) {
+  const chunks = []
+  for await (const chunk of stream) chunks.push(chunk)
+  return Buffer.concat(chunks)
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// detectPayloadFormat
+// ─────────────────────────────────────────────────────────────────────────────
+
+test('detectPayloadFormat - reports a gzipped tar, which is what a Whatsminer streams', (t) => {
+  const format = detectPayloadFormat(zlib.gzipSync(tarBytes()))
+
+  t.is(format.extension, 'tar.gz', 'should name the file .tar.gz')
+  t.is(format.contentType, 'application/gzip', 'should declare gzip')
+})
+
+test('detectPayloadFormat - detects a tar from a truncated stream prefix', (t) => {
+  // The handler only ever sees the first chunk, so the deflate block is cut short.
+  const gzipped = zlib.gzipSync(Buffer.concat([tarBytes(), Buffer.alloc(64 * 1024)]))
+  const format = detectPayloadFormat(gzipped.subarray(0, 512))
+
+  t.is(format.extension, 'tar.gz', 'should still see the tar header')
+})
+
+test('detectPayloadFormat - does not claim tar for a gzip payload that holds none', (t) => {
+  const format = detectPayloadFormat(zlib.gzipSync(Buffer.from('[board0]\npass = 1\n')))
+
+  t.is(format.extension, 'gz', 'should name the file .gz')
+  t.is(format.contentType, 'application/gzip', 'should declare gzip')
+})
+
+test('detectPayloadFormat - reports zip payloads', (t) => {
+  const format = detectPayloadFormat(Buffer.from([0x50, 0x4b, 0x03, 0x04, 0x00]))
+
+  t.is(format.extension, 'zip', 'should name the file .zip')
+  t.is(format.contentType, 'application/zip', 'should declare zip')
+})
+
+test('detectPayloadFormat - falls back to plain text', (t) => {
+  const text = detectPayloadFormat(Buffer.from('[board0]\npass = 1\n'))
+
+  t.is(text.extension, 'log', 'should name a text payload .log')
+  t.is(text.contentType, 'text/plain; charset=utf-8', 'should declare text')
+
+  t.is(detectPayloadFormat(null).extension, 'log', 'should default for an empty stream')
+  t.is(detectPayloadFormat(Buffer.alloc(0)).extension, 'log', 'should default for zero bytes')
+  t.is(detectPayloadFormat(Buffer.from([0x1f])).extension, 'log', 'should default for a single byte')
+})
+
+// ─────────────────────────────────────────────────────────────────────────────
+// peekFirstChunk
+// ─────────────────────────────────────────────────────────────────────────────
+
+test('peekFirstChunk - returns the first chunk', async (t) => {
+  const head = await peekFirstChunk(makeStream(['HEAD', 'MIDDLE', 'TAIL']))
+
+  t.is(head.toString(), 'HEAD', 'should hand back the first chunk')
+})
+
+test('peekFirstChunk - resolves null for an empty stream', async (t) => {
+  t.is(await peekFirstChunk(makeStream([])), null, 'should resolve null')
+})
+
+test('peekFirstChunk - rejects when the stream errors', async (t) => {
+  const stream = new Readable({
+    read (cb) {
+      cb(new Error('ERR_LOG_PEER_TIMEOUT'))
+    }
+  })
+
+  await t.exception(peekFirstChunk(stream), /ERR_LOG_PEER_TIMEOUT/, 'should propagate the error')
+})
+
+// ─────────────────────────────────────────────────────────────────────────────
+// prependChunk
+// ─────────────────────────────────────────────────────────────────────────────
+
+test('prependChunk - re-emits the peeked bytes ahead of the rest', async (t) => {
+  for (const chunks of [['ONLY'], ['HEAD', 'TAIL'], ['A', 'B', 'C']]) {
+    const stream = makeStream(chunks)
+    const head = await peekFirstChunk(stream)
+
+    t.is(
+      (await drain(prependChunk(stream, head))).toString(),
+      chunks.join(''),
+      `should lose no bytes for a ${chunks.length}-chunk payload`
+    )
+  }
+})
+
+test('prependChunk - streams a payload larger than the high-water mark', async (t) => {
+  // stream.unshift() cannot be used for this: it silently drops the head once the peek has
+  // consumed the source's last chunk, which truncated single-chunk logs.
+  const chunks = Array.from({ length: 64 }, (_, i) => Buffer.alloc(64 * 1024, i))
+  const stream = makeStream(chunks)
+  const head = await peekFirstChunk(stream)
+
+  const out = await drain(prependChunk(stream, head))
+
+  t.is(out.length, 64 * 64 * 1024, 'should deliver every byte')
+  t.alike(out, Buffer.concat(chunks), 'should deliver them in order')
+})
+
+test('prependChunk - handles an empty payload', async (t) => {
+  const stream = makeStream([])
+  const head = await peekFirstChunk(stream)
+
+  t.is((await drain(prependChunk(stream, head))).length, 0, 'should end without data')
+})
+
+test('prependChunk - surfaces a source failure that happens after the peek', async (t) => {
+  const stream = makeStream(['HEAD', 'TAIL'])
+  const head = await peekFirstChunk(stream)
+
+  const body = prependChunk(stream, head)
+  stream.destroy(new Error('ERR_LOG_INCOMPLETE'))
+
+  await t.exception(drain(body), /ERR_LOG_INCOMPLETE/, 'should propagate to the consumer')
+})
+
+test('peekFirstChunk - rejects when the source fails before the first chunk arrives', async (t) => {
+  // The connection to the miner can drop mid-transfer; the file leg turns this into a 500
+  // rather than sending a truncated body under a confident content-type.
+  let reads = 0
+  const stream = new Readable({
+    read (cb) {
+      if (reads++ === 0) {
+        this.push(Buffer.from('HEAD'))
+        return cb(null)
+      }
+      cb(new Error('ERR_LOG_INCOMPLETE'))
+    }
+  })
+
+  await t.exception(peekFirstChunk(stream), /ERR_LOG_INCOMPLETE/, 'should reject')
+})
diff --git a/workers/lib/server/handlers/actions.handlers.js b/workers/lib/server/handlers/actions.handlers.js
index 7556249..ea5d8fa 100644
--- a/workers/lib/server/handlers/actions.handlers.js
+++ b/workers/lib/server/handlers/actions.handlers.js
@@ -2,6 +2,7 @@
 
 const { parseJsonQueryParam } = require('../../utils')
 const { ACTIONS_MAX_QUERIES } = require('../../constants')
+const { detectPayloadFormat, peekFirstChunk, prependChunk } = require('../lib/payloadFormat')
 
 async function queryActionsBatch (ctx, req) {
   const payload = {
@@ -208,18 +209,32 @@ async function downloadLogFile (ctx, req, reply) {
     return reply.code(code).send({ error: err.message })
   }
 
+  // The miner decides the payload format — plain text on some models, a gzipped tar of the log
+  // directory on Whatsminers — and the action result carries no format field. Read the leading
+  // bytes so the declared name and type match the payload, then re-emit them ahead of the rest.
+  let head = null
+  try {
+    head = await peekFirstChunk(stream)
+  } catch (err) {
+    stream.destroy()
+    return reply.code(500).send({ error: err.message })
+  }
+
+  const body = prependChunk(stream, head)
+  const { extension, contentType } = detectPayloadFormat(head)
+
   // Set headers only after stream is ready — if set before the try-catch and stream()
-  // throws, the error response would carry application/octet-stream content-type and
-  // Fastify would refuse to serialize the JSON error object.
+  // throws, the error response would carry a binary content-type and Fastify would refuse
+  // to serialize the JSON error object.
   const { safeContentDispositionFilename } = require('../lib/queryUtils')
-  const filename = safeContentDispositionFilename(`miner-log-${meta.minerId || 'unknown'}-${id}.log`)
-  reply.header('Content-Type', 'application/octet-stream')
+  const filename = safeContentDispositionFilename(`miner-log-${meta.minerId || 'unknown'}-${id}.${extension}`)
+  reply.header('Content-Type', contentType)
   reply.header('Content-Disposition', `attachment; filename="${filename}"`)
   reply.header('Content-Length', meta.byteLength)
   reply.header('Cache-Control', 'no-store')
 
   // Fastify pipes a Readable stream directly to the HTTP response — no buffering
-  return reply.send(stream)
+  return reply.send(body)
 }
 
 module.exports = {
diff --git a/workers/lib/server/lib/payloadFormat.js b/workers/lib/server/lib/payloadFormat.js
new file mode 100644
index 0000000..999b875
--- /dev/null
+++ b/workers/lib/server/lib/payloadFormat.js
@@ -0,0 +1,139 @@
+'use strict'
+
+const zlib = require('zlib')
+const { PassThrough } = require('streamx')
+
+// The miner decides what a log download actually contains: some models stream a single
+// plain-text log, Whatsminers stream a gzipped tar of their log directory. The action result
+// carries no format field, so the only reliable source is the payload's leading bytes.
+
+const GZIP_MAGIC = [0x1f, 0x8b]
+const ZIP_MAGIC = [0x50, 0x4b, 0x03, 0x04]
+
+// A POSIX tar header is 512 bytes and carries "ustar" at offset 257.
+const TAR_HEADER_LENGTH = 512
+const TAR_MAGIC = 'ustar'
+const TAR_MAGIC_OFFSET = 257
+
+const TEXT_FORMAT = { extension: 'log', contentType: 'text/plain; charset=utf-8' }
+const GZIP_FORMAT = { extension: 'gz', contentType: 'application/gzip' }
+const TAR_GZ_FORMAT = { extension: 'tar.gz', contentType: 'application/gzip' }
+const ZIP_FORMAT = { extension: 'zip', contentType: 'application/zip' }
+
+function hasMagic (head, magic) {
+  return head.length >= magic.length && magic.every((byte, index) => head[index] === byte)
+}
+
+// Inflates whatever of `head` zlib can manage — the input is a stream prefix, so the deflate
+// block is expected to be truncated. Returns null when it cannot be inflated at all.
+function inflateHead (head) {
+  try {
+    return zlib.gunzipSync(head, { finishFlush: zlib.constants.Z_SYNC_FLUSH })
+  } catch {
+    return null
+  }
+}
+
+function isGzippedTar (head) {
+  const inflated = inflateHead(head)
+  if (!inflated || inflated.length < TAR_HEADER_LENGTH) return false
+
+  return inflated.toString('latin1', TAR_MAGIC_OFFSET, TAR_MAGIC_OFFSET + TAR_MAGIC.length) === TAR_MAGIC
+}
+
+/**
+ * Format implied by a payload's leading bytes. Falls back to plain text, and never claims 'tar'
+ * without having seen a tar header, so the declared name always matches the bytes.
+ *
+ * @param {Buffer|null} head  First chunk of the payload (null for an empty stream)
+ * @returns {{ extension: string, contentType: string }}
+ */
+function detectPayloadFormat (head) {
+  if (!head || !head.length) return TEXT_FORMAT
+
+  if (hasMagic(head, GZIP_MAGIC)) {
+    return isGzippedTar(head) ? TAR_GZ_FORMAT : GZIP_FORMAT
+  }
+
+  if (hasMagic(head, ZIP_MAGIC)) return ZIP_FORMAT
+
+  return TEXT_FORMAT
+}
+
+/**
+ * Reads the first chunk of a stream. Pair it with `prependChunk` to hand the bytes back —
+ * `stream.unshift` silently drops them when the peek consumed the stream's last chunk, which
+ * would truncate a small log. Resolves null when the stream ends without producing data.
+ *
+ * @param {import('streamx').Readable} stream
+ * @returns {Promise<Buffer|null>}
+ */
+function peekFirstChunk (stream) {
+  return new Promise((resolve, reject) => {
+    let settled = false
+
+    // Stays attached for the life of the stream: an 'error' with no listener is thrown by
+    // EventEmitter, and the source can fail in the gap between this peek and the pipe.
+    const onError = (err) => {
+      if (settled) return
+      settled = true
+      detach()
+      reject(err)
+    }
+
+    const detach = () => {
+      stream.off('readable', onReadable)
+      stream.off('end', onEnd)
+    }
+    const settle = (chunk) => {
+      if (settled) return
+      settled = true
+      detach()
+      resolve(chunk === undefined ? null : chunk)
+    }
+    const onReadable = () => settle(stream.read())
+    const onEnd = () => settle(null)
+
+    stream.on('error', onError)
+
+    const immediate = stream.read()
+    if (immediate !== null && immediate !== undefined) return settle(immediate)
+
+    stream.on('readable', onReadable)
+    stream.on('end', onEnd)
+  })
+}
+
+/**
+ * Re-emits a peeked chunk ahead of the rest of the source, so the consumer sees the payload
+ * byte-for-byte. Honours the consumer's backpressure — nothing beyond one chunk is buffered.
+ *
+ * @param {import('streamx').Readable} source  Stream already advanced past `head`
+ * @param {Buffer|null} head
+ * @returns {import('streamx').Readable}
+ */
+function prependChunk (source, head) {
+  const out = new PassThrough()
+
+  const write = async (chunk) => {
+    if (out.write(chunk) === false) {
+      await new Promise((resolve) => out.once('drain', resolve))
+    }
+  }
+
+  const pump = async () => {
+    if (head) await write(head)
+    for await (const chunk of source) await write(chunk)
+    out.end()
+  }
+
+  pump().catch((err) => out.destroy(err))
+
+  return out
+}
+
+module.exports = {
+  detectPayloadFormat,
+  peekFirstChunk,
+  prependChunk
+}

From 4d30f0be6faedae53961ccdca35acd3529069611 Mon Sep 17 00:00:00 2001
From: arif-dewi <11630380+arif-dewi@users.noreply.github.com>
Date: Thu, 27 Aug 2026 13:06:34 +0400
Subject: [PATCH 2/2] fix(miner-logs): tie the log stream lifecycles and bound
 the format probe
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Review follow-ups on the payload-format detection:

- A client aborting mid-download left the transfer open: fastify destroys the
  response stream, but nothing destroyed the hypercore source, and the pump
  parked on a drain that would never fire. The two now share a lifecycle —
  destroying the output destroys the source, and a write to a gone consumer
  stops the pump instead of hanging it. source.pipe() would tie them for free
  but cannot be used here: the peek may already have consumed the source's
  last chunk, and piping an ended stream never ends the destination (that is
  what made the earlier unshift version drop single-chunk logs).

- `head` is device-supplied and deflate expands up to ~1000x, so probing a
  whole chunk could inflate ~67MB synchronously per request. Only the first
  4096 bytes are inflated now — 4.2MB and 1.35ms on a 64MB-of-zeros payload,
  with tar detection unchanged since it needs 512 output bytes.
  maxOutputLength is deliberately not used: it throws, which would demote a
  real .tar.gz to .gz.

- The output stream is Node's stream.PassThrough rather than streamx, which
  matches the convention already used in export/serializers.js and drops the
  runtime dependency entirely. The tests keep streamx.Readable, since that is
  what hypercore.createByteStream() returns and its peek semantics differ, so
  streamx is now declared as a devDependency instead of resolving through
  hoisting from a transitive dep.

Both fixes have regression coverage: removing the lifecycle handler fails the
abort test on its timeout, and the inflate bound is asserted on output size
rather than timing.
---
 package-lock.json                       |  3 +-
 package.json                            |  3 +-
 tests/unit/lib/payloadFormat.test.js    | 47 +++++++++++++++++++++
 workers/lib/server/lib/payloadFormat.js | 56 ++++++++++++++++++++-----
 4 files changed, 96 insertions(+), 13 deletions(-)

diff --git a/package-lock.json b/package-lock.json
index 82b058b..34670ef 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -33,7 +33,8 @@
         "@tetherto/tether-svc-test-helper": "git+https://github.com/tetherto/tether-svc-test-helper.git#v1.0.0",
         "brittle": "3.18.0",
         "http-server": "14.1.1",
-        "standard": "17.1.2"
+        "standard": "17.1.2",
+        "streamx": "2.28.0"
       },
       "engines": {
         "node": ">=24"
diff --git a/package.json b/package.json
index 0bbcd90..1dc6e39 100644
--- a/package.json
+++ b/package.json
@@ -63,7 +63,8 @@
     "@tetherto/tether-svc-test-helper": "git+https://github.com/tetherto/tether-svc-test-helper.git#v1.0.0",
     "brittle": "3.18.0",
     "http-server": "14.1.1",
-    "standard": "17.1.2"
+    "standard": "17.1.2",
+    "streamx": "2.28.0"
   },
   "overrides": {
     "@tootallnate/once": "3.0.1",
diff --git a/tests/unit/lib/payloadFormat.test.js b/tests/unit/lib/payloadFormat.test.js
index a951fcd..9f2cf93 100644
--- a/tests/unit/lib/payloadFormat.test.js
+++ b/tests/unit/lib/payloadFormat.test.js
@@ -5,6 +5,7 @@ const { Readable } = require('streamx')
 const test = require('brittle')
 
 const {
+  inflateHead,
   detectPayloadFormat,
   peekFirstChunk,
   prependChunk
@@ -170,3 +171,49 @@ test('peekFirstChunk - rejects when the source fails before the first chunk arri
 
   await t.exception(peekFirstChunk(stream), /ERR_LOG_INCOMPLETE/, 'should reject')
 })
+
+// ─────────────────────────────────────────────────────────────────────────────
+// inflateHead — bounded work on device-supplied bytes
+// ─────────────────────────────────────────────────────────────────────────────
+
+test('inflateHead - inflates only a bounded slice of a highly compressible chunk', (t) => {
+  // `head` comes from the miner and deflate expands up to ~1000x: inflating a whole 64 KB chunk
+  // of this shape would allocate ~67 MB synchronously per request.
+  const bomb = zlib.gzipSync(Buffer.alloc(64 * 1024 * 1024))
+  const inflated = inflateHead(bomb)
+
+  t.ok(inflated.length >= 512, 'should still yield enough output to read a tar header')
+  t.ok(
+    inflated.length < 8 * 1024 * 1024,
+    `should not inflate the whole payload (got ${inflated.length} bytes)`
+  )
+  t.is(detectPayloadFormat(bomb).extension, 'gz', 'should still classify the payload')
+})
+
+test('inflateHead - still sees the tar header of a real archive', (t) => {
+  t.is(detectPayloadFormat(zlib.gzipSync(tarBytes())).extension, 'tar.gz', 'should detect tar')
+})
+
+// ─────────────────────────────────────────────────────────────────────────────
+// prependChunk — lifecycle
+// ─────────────────────────────────────────────────────────────────────────────
+
+test('prependChunk - destroys the source when the consumer goes away', async (t) => {
+  // Fastify destroys the response stream when a client aborts mid-download. Nothing else would
+  // close the P2P transfer from the miner, and the pump would park on a drain that never fires.
+  const source = makeStream(Array.from({ length: 64 }, () => Buffer.alloc(64 * 1024, 7)))
+  const head = await peekFirstChunk(source)
+  const body = prependChunk(source, head)
+
+  await new Promise((resolve) => body.once('readable', resolve))
+  body.destroy()
+
+  await t.execution(
+    Promise.race([
+      new Promise((resolve) => source.once('close', resolve)),
+      new Promise((resolve, reject) => setTimeout(() => reject(new Error('source never closed')), 2000))
+    ]),
+    'should destroy the source rather than leave the transfer open'
+  )
+  t.ok(source.destroyed, 'source should be destroyed')
+})
diff --git a/workers/lib/server/lib/payloadFormat.js b/workers/lib/server/lib/payloadFormat.js
index 999b875..2bb353c 100644
--- a/workers/lib/server/lib/payloadFormat.js
+++ b/workers/lib/server/lib/payloadFormat.js
@@ -1,7 +1,7 @@
 'use strict'
 
 const zlib = require('zlib')
-const { PassThrough } = require('streamx')
+const { PassThrough } = require('stream')
 
 // The miner decides what a log download actually contains: some models stream a single
 // plain-text log, Whatsminers stream a gzipped tar of their log directory. The action result
@@ -10,6 +10,10 @@ const { PassThrough } = require('streamx')
 const GZIP_MAGIC = [0x1f, 0x8b]
 const ZIP_MAGIC = [0x50, 0x4b, 0x03, 0x04]
 
+// `head` is device-supplied and deflate expands up to ~1000x, so only a bounded slice of it is
+// ever inflated — detection needs 512 output bytes, not the whole chunk.
+const GZIP_INFLATE_INPUT_LIMIT = 4096
+
 // A POSIX tar header is 512 bytes and carries "ustar" at offset 257.
 const TAR_HEADER_LENGTH = 512
 const TAR_MAGIC = 'ustar'
@@ -24,11 +28,15 @@ function hasMagic (head, magic) {
   return head.length >= magic.length && magic.every((byte, index) => head[index] === byte)
 }
 
-// Inflates whatever of `head` zlib can manage — the input is a stream prefix, so the deflate
-// block is expected to be truncated. Returns null when it cannot be inflated at all.
+// Inflates whatever zlib can manage of the first GZIP_INFLATE_INPUT_LIMIT bytes — the input is a
+// stream prefix, so the deflate block is expected to be truncated. Returns null when it cannot be
+// inflated at all. `maxOutputLength` is deliberately not used: it throws, which would demote a
+// real .tar.gz to .gz.
 function inflateHead (head) {
   try {
-    return zlib.gunzipSync(head, { finishFlush: zlib.constants.Z_SYNC_FLUSH })
+    return zlib.gunzipSync(head.subarray(0, GZIP_INFLATE_INPUT_LIMIT), {
+      finishFlush: zlib.constants.Z_SYNC_FLUSH
+    })
   } catch {
     return null
   }
@@ -108,22 +116,47 @@ function peekFirstChunk (stream) {
  * Re-emits a peeked chunk ahead of the rest of the source, so the consumer sees the payload
  * byte-for-byte. Honours the consumer's backpressure — nothing beyond one chunk is buffered.
  *
- * @param {import('streamx').Readable} source  Stream already advanced past `head`
+ * The two streams share a lifecycle: if the consumer goes away (fastify destroys the response
+ * stream when a client aborts mid-download) the source is destroyed too, so the P2P transfer from
+ * the miner is not left open with the pump parked on a drain that will never fire. `source.pipe()`
+ * would tie them for free but cannot be used here — the peek may already have consumed the
+ * source's last chunk, and piping an ended stream never ends the destination.
+ *
+ * @param {import('stream').Readable} source  Stream already advanced past `head`
  * @param {Buffer|null} head
- * @returns {import('streamx').Readable}
+ * @returns {import('stream').Readable}
  */
 function prependChunk (source, head) {
   const out = new PassThrough()
 
+  out.on('close', () => source.destroy())
+
+  // Resolves false once the consumer is gone, which stops the pump instead of hanging it.
   const write = async (chunk) => {
-    if (out.write(chunk) === false) {
-      await new Promise((resolve) => out.once('drain', resolve))
-    }
+    if (out.destroyed) return false
+    if (out.write(chunk) !== false) return true
+
+    return new Promise((resolve) => {
+      const settle = (delivered) => {
+        out.off('drain', onDrain)
+        out.off('close', onClose)
+        resolve(delivered)
+      }
+      const onDrain = () => settle(true)
+      const onClose = () => settle(false)
+
+      out.on('drain', onDrain)
+      out.on('close', onClose)
+    })
   }
 
   const pump = async () => {
-    if (head) await write(head)
-    for await (const chunk of source) await write(chunk)
+    if (head && !(await write(head))) return
+
+    for await (const chunk of source) {
+      if (!(await write(chunk))) return
+    }
+
     out.end()
   }
 
@@ -133,6 +166,7 @@ function prependChunk (source, head) {
 }
 
 module.exports = {
+  inflateHead,
   detectPayloadFormat,
   peekFirstChunk,
   prependChunk