SYM-2983: Resume transfer of large batches using HTTP Range headers

symmetric-assemble/src/asciidoc/advanced-topics.ad

@@ -486,6 +486,9 @@ is uploaded as an HTTP multi-part attachment. Outgoing zip files are written and
 Incoming zip files are staged in the filesync_incoming staging directory by source node id. The filesync_incoming/{node_id} staging 
 directory is cleared out before each subsequent delivery of files.
 
+A file sync pull may bundle more than one batch's zip into a single response, and an interrupted pull can resume from where it left
+off instead of re-downloading everything; see <<Resumable Batch Transfer>> for details.
+
 The acknowledgement of a batch happens the same way it is acknowledged in database synchronization. The client responds with an 
 acknowledgement as part of the response during a file push or pull.
 

symmetric-assemble/src/asciidoc/advanced/transport-manager.ad

@@ -61,3 +61,55 @@ http.transport.manager.class=com.example.MyHttpTransportManager
 
 This applies to both the `http` and `hybrid` transport types, since the hybrid transport uses an HTTP
 transport manager internally for remote communication.
+
+==== Resumable Batch Transfer
+
+When a batch pull is interrupted partway through (for example, by a dropped connection through a proxy or
+load balancer), SymmetricDS can resume the transfer from where it left off instead of re-downloading the
+whole batch. This applies to both table (CSV) data batches and file sync (ZIP) batches, and only to the
+pull direction; a push always sends the full batch.
+
+Resumable transfer is controlled by the `sync.http.resume.enabled` parameter, which defaults to `true`.
+
+[source, properties]
+----
+sync.http.resume.enabled=true
+----
+
+Resume only takes effect for batches that are staged to disk, so it also requires `stream.to.file.enabled=true`
+and a batch size over the `stream.to.file.threshold.bytes` threshold; smaller batches are streamed from memory
+and always re-sent in full.
+
+.How it works
+* Each staged batch is identified by an ETag made up of the batch's staging version, the time its staged file
+was generated, and its final size. The client caches this ETag along with how much of the batch it has already
+received whenever a pull attempt fails partway through.
+* On the next pull attempt for that node, the client sends the cached ETag in an `If-ETag` header and how much
+it already has in a `Range` header, along with the ID of the batch to resume. Table (CSV) batches use
+`Range: chars=<count>-`, a count of decoded characters, since the staged content is read and written as text;
+file sync (ZIP) batches use `Range: bytes=<count>-`, a count of raw bytes, since the staged content is copied
+as-is.
+* If the server still has that exact staged batch (matched by ETag) available, it responds with
+`206 Partial Content`, an `ETag` header, and a `Content-Range` header, and streams only the remainder.
+* If the staged batch is missing or its ETag no longer matches (for example, it was purged or re-extracted),
+the server falls back to a normal `200` response with the full batch, and the client discards its partial
+copy and starts over.
+
+Older peers (prior to version 3.18) simply do not send the resume request headers, so they always receive
+full-batch behavior; the resume mechanism only activates when both sides support it.
+
+These `Range`/`If-ETag`/`Accept-Ranges` headers are a private convention between SymmetricDS nodes, not a
+standard, cache-aware `If-Range` exchange - a node never validates a range against a shared intermediary
+cache. Because of this, a proxy, load balancer, or CDN that inspects and tries to interpret or satisfy HTTP
+range requests on its own (rather than passing them through unmodified) can corrupt a resumed transfer; make
+sure any such intermediary in the sync path is configured to forward these headers untouched rather than act
+on them.
+
+.File sync bundling
+For file sync pulls specifically, each batch is staged as its own independent zip file, the same way table
+batches already are, so that any one batch can be resumed without affecting the others. To keep bundling
+multiple batches into a single pull response (subject to `transport.max.bytes.to.sync`), SymmetricDS wraps
+each batch's zip in a small header — batch ID, byte length, and ETag — one after another in the response
+body. A `FileSync-Format` response header tells the client that this bundling format was used; it is only
+sent when the target node is version 3.18 or later, since older clients only know how to unzip a single,
+unwrapped zip file per response. When the header is absent, the response is a single, unwrapped zip file.

symmetric-core/src/main/java/org/jumpmind/symmetric/common/ParameterConstants.java

@@ -266,6 +266,7 @@ private ParameterConstants() {
     public static final String TRANSPORT_HTTP_SESSION_EXPIRE_SECONDS = "http.session.expire.seconds";
     public static final String TRANSPORT_HTTP_SESSION_MAX_COUNT = "http.session.max.count";
     public static final String TRANSPORT_HTTP_USE_HEADER_SECURITY_TOKEN = "http.use.header.security.token";
+    public static final String TRANSPORT_HTTP_RESUME_ENABLED = "sync.http.resume.enabled";
     public static final String TRANSPORT_TYPE = "transport.type";
     public static final String TRANSPORT_MAX_BYTES_TO_SYNC = "transport.max.bytes.to.sync";
     public static final String TRANSPORT_MAX_FORM_KEYS = "transport.max.form.keys";

symmetric-core/src/main/java/org/jumpmind/symmetric/extract/CountingSkippingWriter.java

@@ -0,0 +1,75 @@
+/**
+ * Licensed to JumpMind Inc under one or more contributor
+ * license agreements.  See the NOTICE file distributed
+ * with this work for additional information regarding
+ * copyright ownership.  JumpMind Inc licenses this file
+ * to you under the GNU Affero General Public License, version 3.0 (AGPLv3)
+ * (the "License"); you may not use this file except in compliance
+ * with the License.
+ *
+ * You should have received a copy of the GNU Affero General Public License,
+ * version 3.0 (AGPLv3) along with this library; if not, see
+ * <http://www.gnu.org/licenses/>.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.jumpmind.symmetric.extract;
+
+import java.io.IOException;
+import java.io.Writer;
+
+/**
+ * Wraps a destination {@link Writer}, discarding the first {@code skipCount} characters written to it and forwarding the rest, while counting the total number
+ * of characters seen (skipped plus forwarded). This lets a single deterministic write pass serve both a full batch resend ({@code skipCount == 0}) and a
+ * resumed, partial send ({@code skipCount > 0}) starting from the same point in the stream.
+ * <p>
+ * The count is in decoded characters of the underlying CSV text stream, not raw network bytes: the staged resource is read and written through
+ * {@link java.io.Reader}/{@link java.io.Writer}, not {@link java.io.InputStream}/{@link java.io.OutputStream}, so an HTTP Range/Content-Range value used with
+ * this class must agree on that same unit on both the client and server side. Since both sides read the exact same staged, UTF-8 file deterministically, this
+ * is internally consistent even though it is not a literal byte offset per RFC 9110 Range semantics.
+ */
+public class CountingSkippingWriter extends Writer {
+    private final Writer delegate;
+    private final long skipCount;
+    private long totalCount;
+
+    public CountingSkippingWriter(Writer delegate, long skipCount) {
+        this.delegate = delegate;
+        this.skipCount = skipCount;
+    }
+
+    public long getTotalCount() {
+        return totalCount;
+    }
+
+    @Override
+    public void write(char[] cbuf, int off, int len) throws IOException {
+        int writeOff = off;
+        int writeLen = len;
+        if (totalCount < skipCount) {
+            long remainingToSkip = skipCount - totalCount;
+            int skipInThisChunk = (int) Math.min(remainingToSkip, len);
+            writeOff = off + skipInThisChunk;
+            writeLen = len - skipInThisChunk;
+        }
+        if (writeLen > 0) {
+            delegate.write(cbuf, writeOff, writeLen);
+        }
+        totalCount += len;
+    }
+
+    @Override
+    public void flush() throws IOException {
+        delegate.flush();
+    }
+
+    @Override
+    public void close() throws IOException {
+        delegate.close();
+    }
+}

symmetric-core/src/main/java/org/jumpmind/symmetric/file/FileSyncBatchEnvelope.java

@@ -0,0 +1,92 @@
+/**
+ * Licensed to JumpMind Inc under one or more contributor
+ * license agreements.  See the NOTICE file distributed
+ * with this work for additional information regarding
+ * copyright ownership.  JumpMind Inc licenses this file
+ * to you under the GNU Affero General Public License, version 3.0 (AGPLv3)
+ * (the "License"); you may not use this file except in compliance
+ * with the License.
+ *
+ * You should have received a copy of the GNU Affero General Public License,
+ * version 3.0 (AGPLv3) along with this library; if not, see
+ * <http://www.gnu.org/licenses/>.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.jumpmind.symmetric.file;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+
+import org.jumpmind.symmetric.io.stage.StagedResourceETag;
+
+/**
+ * A lightweight header written immediately before each batch's complete, independently-staged zip bytes in a {@code FileSync-Format}-tagged pull response, so
+ * several batches can be bundled into one response while still letting the reader know exactly where one batch's zip ends and the next one's header begins — a
+ * purely length-based framing, no entry-by-entry inspection required.
+ * <p>
+ * Wire shape is one UTF-8 text line, {@code <batchId>,<zipByteLength>,<etagJson>}, followed by exactly {@code zipByteLength} raw zip bytes. The ETag JSON
+ * itself may contain commas, so only the first two commas are treated as delimiters; everything after the second comma is taken verbatim as the ETag JSON.
+ */
+public class FileSyncBatchEnvelope {
+    private final long batchId;
+    private final long length;
+    private final StagedResourceETag etag;
+
+    public FileSyncBatchEnvelope(long batchId, long length, StagedResourceETag etag) {
+        this.batchId = batchId;
+        this.length = length;
+        this.etag = etag;
+    }
+
+    public long getBatchId() {
+        return batchId;
+    }
+
+    public long getLength() {
+        return length;
+    }
+
+    public StagedResourceETag getEtag() {
+        return etag;
+    }
+
+    public static void writeHeader(OutputStream out, long batchId, long length, StagedResourceETag etag) throws IOException {
+        String line = batchId + "," + length + "," + etag.toJson() + "\n";
+        out.write(line.getBytes(StandardCharsets.UTF_8));
+    }
+
+    /**
+     * Reads one envelope header line from {@code in}, one byte at a time so as to never consume bytes past the header's trailing newline — the caller must read
+     * exactly {@link #getLength()} bytes immediately afterward, so any over-read here would corrupt the following batch's zip content.
+     *
+     * @return the parsed header, or {@code null} at a clean end of stream (no more batches follow)
+     */
+    public static FileSyncBatchEnvelope readHeader(InputStream in) throws IOException {
+        StringBuilder line = new StringBuilder();
+        int b;
+        while ((b = in.read()) != -1 && b != '\n') {
+            line.append((char) b);
+        }
+        if (b == -1 && line.isEmpty()) {
+            return null;
+        }
+        String headerLine = line.toString();
+        int firstComma = headerLine.indexOf(',');
+        int secondComma = firstComma < 0 ? -1 : headerLine.indexOf(',', firstComma + 1);
+        if (firstComma < 0 || secondComma < 0) {
+            throw new IOException("Malformed file sync envelope header: " + headerLine);
+        }
+        long batchId = Long.parseLong(headerLine.substring(0, firstComma));
+        long length = Long.parseLong(headerLine.substring(firstComma + 1, secondComma));
+        StagedResourceETag etag = StagedResourceETag.fromJson(headerLine.substring(secondComma + 1));
+        return new FileSyncBatchEnvelope(batchId, length, etag);
+    }
+}

symmetric-core/src/main/java/org/jumpmind/symmetric/file/FileSyncPullResult.java

@@ -0,0 +1,150 @@
+/**
+ * Licensed to JumpMind Inc under one or more contributor
+ * license agreements.  See the NOTICE file distributed
+ * with this work for additional information regarding
+ * copyright ownership.  JumpMind Inc licenses this file
+ * to you under the GNU Affero General Public License, version 3.0 (AGPLv3)
+ * (the "License"); you may not use this file except in compliance
+ * with the License.
+ *
+ * You should have received a copy of the GNU Affero General Public License,
+ * version 3.0 (AGPLv3) along with this library; if not, see
+ * <http://www.gnu.org/licenses/>.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.jumpmind.symmetric.file;
+
+import java.util.List;
+
+import org.jumpmind.symmetric.io.stage.IStagedResource;
+import org.jumpmind.symmetric.io.stage.StagedResourceETag;
+import org.jumpmind.symmetric.model.OutgoingBatch;
+
+/**
+ * The outcome of {@link org.jumpmind.symmetric.service.IFileSyncService#prepareFilesForPull}, carrying both what the servlet handler needs to set response
+ * headers/status, and what {@link org.jumpmind.symmetric.service.IFileSyncService#writeFilesForPull} needs to stream the previously-staged bytes afterward.
+ * Split from a single combined call so the handler can set headers on the servlet response <em>before</em> any bytes are written to it - setting a header on an
+ * already-committed response is a silent no-op, which previously meant the {@code FileSync-Format} header was never actually sent to the client.
+ * <p>
+ * {@code resumeEtag} is non-null only when this response served (or attempted to serve) exactly one specific, previously-interrupted batch by request; it is
+ * {@code null} for a normal, non-resume pull. {@code allRequestedBatches} is only meaningful for a normal (non-resume) pull - it is the full candidate list
+ * {@code batches} was selected from, needed by {@link org.jumpmind.symmetric.service.IFileSyncService#writeFilesForPull} to mark them loaded.
+ */
+public class FileSyncPullResult {
+    private final List<OutgoingBatch> batches;
+    private final List<OutgoingBatch> allRequestedBatches;
+    private final List<IStagedResource> stagedResources;
+    private final boolean isEnvelopeFormatUsed;
+    private final boolean isPartialContent;
+    private final StagedResourceETag resumeEtag;
+    private final long totalSize;
+    private final long skipCount;
+
+    private FileSyncPullResult(Builder builder) {
+        this.batches = builder.batches;
+        this.allRequestedBatches = builder.allRequestedBatches;
+        this.stagedResources = builder.stagedResources;
+        this.isEnvelopeFormatUsed = builder.isEnvelopeFormatUsed;
+        this.isPartialContent = builder.isPartialContent;
+        this.resumeEtag = builder.resumeEtag;
+        this.totalSize = builder.totalSize;
+        this.skipCount = builder.skipCount;
+    }
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    public List<OutgoingBatch> getBatches() {
+        return batches;
+    }
+
+    public List<OutgoingBatch> getAllRequestedBatches() {
+        return allRequestedBatches;
+    }
+
+    public List<IStagedResource> getStagedResources() {
+        return stagedResources;
+    }
+
+    public boolean isEnvelopeFormatUsed() {
+        return isEnvelopeFormatUsed;
+    }
+
+    public boolean isPartialContent() {
+        return isPartialContent;
+    }
+
+    public StagedResourceETag getResumeEtag() {
+        return resumeEtag;
+    }
+
+    public long getTotalSize() {
+        return totalSize;
+    }
+
+    public long getSkipCount() {
+        return skipCount;
+    }
+
+    public static class Builder {
+        private List<OutgoingBatch> batches;
+        private List<OutgoingBatch> allRequestedBatches;
+        private List<IStagedResource> stagedResources;
+        private boolean isEnvelopeFormatUsed;
+        private boolean isPartialContent;
+        private StagedResourceETag resumeEtag;
+        private long totalSize;
+        private long skipCount;
+
+        public Builder batches(List<OutgoingBatch> batches) {
+            this.batches = batches;
+            return this;
+        }
+
+        public Builder allRequestedBatches(List<OutgoingBatch> allRequestedBatches) {
+            this.allRequestedBatches = allRequestedBatches;
+            return this;
+        }
+
+        public Builder stagedResources(List<IStagedResource> stagedResources) {
+            this.stagedResources = stagedResources;
+            return this;
+        }
+
+        public Builder envelopeFormatUsed(boolean isEnvelopeFormatUsed) {
+            this.isEnvelopeFormatUsed = isEnvelopeFormatUsed;
+            return this;
+        }
+
+        public Builder partialContent(boolean isPartialContent) {
+            this.isPartialContent = isPartialContent;
+            return this;
+        }
+
+        public Builder resumeEtag(StagedResourceETag resumeEtag) {
+            this.resumeEtag = resumeEtag;
+            return this;
+        }
+
+        public Builder totalSize(long totalSize) {
+            this.totalSize = totalSize;
+            return this;
+        }
+
+        public Builder skipCount(long skipCount) {
+            this.skipCount = skipCount;
+            return this;
+        }
+
+        public FileSyncPullResult build() {
+            return new FileSyncPullResult(this);
+        }
+    }
+}