tgz: underflow/overflow proxy stream

In tar, the entry header contains the file size. If we don't provide it those many bytes, the tar will become corrupt
Linux provides no guarantee of how many bytes can be read from a file. This is the case with sqlite and log files
which are accessed by other processes when tar is in action. This class handles overflow and underflow

src/backupformat/tgz.js

@@ -18,6 +18,7 @@ const assert = require('assert'),
     safe = require('safetydance'),
     storage = require('../storage.js'),
     stream = require('stream/promises'),
+    { Transform } = require('node:stream'),
     tar = require('tar-stream'),
     zlib = require('zlib');
 
@@ -34,6 +35,42 @@ function getBackupFilePath(backupConfig, remotePath) {
     return path.join(rootPath, remotePath + fileType);
 }
 
+// In tar, the entry header contains the file size. If we don't provide it those many bytes, the tar will become corrupt
+// Linux provides no guarantee of how many bytes can be read from a file. This is the case with sqlite and log files
+// which are accessed by other processes when tar is in action. This class handles overflow and underflow
+class EnsureFileSizeStream extends Transform {
+    constructor(options) {
+        super(options);
+        this._remaining = options.size;
+        this._name = options.name;
+    }
+
+    _transform(chunk, encoding, callback) {
+        if (this._remaining <= 0) {
+            debug(`EnsureFileSizeStream: ${this._name} dropping ${chunk.length} bytes`);
+            return callback(null);
+        }
+
+        if (this._remaining - chunk.length < 0) {
+            debug(`EnsureFileSizeStream: ${this._name} dropping extra ${chunk.length - this._remaining} bytes`);
+            chunk = chunk.subarray(0, this._remaining);
+            this._remaining = 0;
+        } else {
+            this._remaining -= chunk.length;
+        }
+
+        callback(null, chunk);
+    }
+
+    _flush(callback) {
+        if (this._remaining > 0) {
+            debug(`EnsureFileSizeStream: ${this._name} injecting ${this._remaining} bytes`);
+            this.push(Buffer.alloc(this._remaining, 0));
+        }
+        callback();
+    }
+}
+
 function addEntryToPack(pack, header, options) {
     assert.strictEqual(typeof pack, 'object');
     assert.strictEqual(typeof header, 'object');
@@ -52,7 +89,10 @@ function addEntryToPack(pack, header, options) {
 
         if (!packEntry) return reject(new BoxError(BoxError.FS_ERROR, `Failed to add ${header.name}: ${safe.error.message}`));
 
-        if (options?.input) safe(stream.pipeline(options.input, packEntry), { debug }); // background. rely on pack.entry callback for promise completion
+        if (options?.input) {
+            const ensureFileSizeStream = new EnsureFileSizeStream({ name: header.name, size: header.size });
+            safe(stream.pipeline(options.input, ensureFileSizeStream, packEntry), { debug }); // background. rely on pack.entry callback for promise completion
+        }
     });
 }