feat: live stream progress — ZstdCompressStream.progress()

ZstdCompressStream.progress() returns a ZstdFrameProgression snapshot
(ingested, consumed, produced, flushed, currentJobId, activeWorkers) via
ZSTD_getFrameProgression (a struct returned by value through FFM) — byte
counters for reporting on a long compression stream.

Tests: counters are zero on a fresh stream; after a complete END step consumed
== ingested == input length and flushed == produced; activeWorkers is 0 on the
single-threaded build.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: dfa1

zstd/src/main/java/io/github/dfa1/zstd/Bindings.java

@@ -166,6 +166,14 @@ final class Bindings {
     static final MethodHandle DPARAM_GET_BOUNDS =
             NativeLibrary.lookup("ZSTD_dParam_getBounds", FunctionDescriptor.of(BOUNDS_LAYOUT, JAVA_INT));
 
+    // ZSTD_frameProgression { u64 ingested, consumed, produced, flushed; u32 currentJobID, nbActiveWorkers; }
+    private static final MemoryLayout FRAME_PROGRESSION_LAYOUT =
+            MemoryLayout.structLayout(JAVA_LONG, JAVA_LONG, JAVA_LONG, JAVA_LONG, JAVA_INT, JAVA_INT);
+    // ZSTD_frameProgression ZSTD_getFrameProgression(const ZSTD_CCtx*) — returned by value
+    static final MethodHandle GET_FRAME_PROGRESSION =
+            NativeLibrary.lookup("ZSTD_getFrameProgression",
+                    FunctionDescriptor.of(FRAME_PROGRESSION_LAYOUT, ADDRESS));
+
     // size_t ZSTD_compressStream2(ZSTD_CCtx*, ZSTD_outBuffer*, ZSTD_inBuffer*, ZSTD_EndDirective)
     static final MethodHandle COMPRESS_STREAM2 =
             NativeLibrary.lookup("ZSTD_compressStream2",

zstd/src/main/java/io/github/dfa1/zstd/ZstdCompressStream.java

@@ -2,6 +2,10 @@
 
 import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
+import java.lang.foreign.SegmentAllocator;
+
+import static java.lang.foreign.ValueLayout.JAVA_INT;
+import static java.lang.foreign.ValueLayout.JAVA_LONG;
 
 /// A low-level, zero-copy streaming compressor driven with native
 /// {@link MemorySegment} buffers — no heap `byte[]` bounce.
@@ -89,6 +93,26 @@ public ZstdStreamResult compress(MemorySegment dst, MemorySegment src, ZstdEndDi
         return new ZstdStreamResult(in.pos(), out.pos(), remaining);
     }
 
+    /// A live snapshot of how much this stream has ingested, compressed, produced,
+    /// and flushed so far — useful for progress reporting on a long stream.
+    ///
+    /// @return the current frame progression
+    public ZstdFrameProgression progress() {
+        try (Arena scratch = Arena.ofConfined()) {
+            MemorySegment p = (MemorySegment) Bindings.GET_FRAME_PROGRESSION.invokeExact(
+                    (SegmentAllocator) scratch, ptr());
+            return new ZstdFrameProgression(
+                    p.get(JAVA_LONG, 0),    // ingested
+                    p.get(JAVA_LONG, 8),    // consumed
+                    p.get(JAVA_LONG, 16),   // produced
+                    p.get(JAVA_LONG, 24),   // flushed
+                    p.get(JAVA_INT, 32),    // currentJobID
+                    p.get(JAVA_INT, 36));   // nbActiveWorkers
+        } catch (Throwable t) {
+            throw rethrow(t);
+        }
+    }
+
     @Override
     protected void tryClose(MemorySegment ptr) throws Throwable {
         try {

zstd/src/main/java/io/github/dfa1/zstd/ZstdFrameProgression.java

@@ -0,0 +1,19 @@
+package io.github.dfa1.zstd;
+
+/// A live snapshot of a streaming compressor's progress, from
+/// `ZSTD_getFrameProgression`.
+///
+/// @param ingested      input bytes read and buffered so far
+/// @param consumed      input bytes actually compressed so far
+/// @param produced      output bytes generated so far, including still-buffered
+/// @param flushed       output bytes flushed out so far
+/// @param currentJobId  current multithreading job id (0 when single-threaded)
+/// @param activeWorkers number of worker threads currently active (0 when single-threaded)
+public record ZstdFrameProgression(
+        long ingested,
+        long consumed,
+        long produced,
+        long flushed,
+        int currentJobId,
+        int activeWorkers) {
+}

zstd/src/test/java/io/github/dfa1/zstd/ZstdSegmentStreamTest.java

@@ -88,6 +88,33 @@ private byte[] drive(byte[] input, int chunk, boolean compress) {
         }
     }
 
+    @Nested
+    class Progress {
+
+        @Test
+        void tracksByteCounters() {
+            byte[] original = "progress payload ".repeat(500).getBytes(StandardCharsets.UTF_8);
+            try (Arena arena = Arena.ofConfined();
+                 ZstdCompressStream cs = new ZstdCompressStream()) {
+
+                MemorySegment src = segment(arena, original);
+                MemorySegment dst = arena.allocate(Zstd.compressBound(original.length));
+
+                // fresh stream: nothing moved yet
+                assertThat(cs.progress().consumed()).isZero();
+
+                ZstdStreamResult r = cs.compress(dst, src, ZstdEndDirective.END);
+
+                // after a complete END step the counters reflect the whole frame
+                ZstdFrameProgression p = cs.progress();
+                assertThat(p.consumed()).isEqualTo(original.length);
+                assertThat(p.ingested()).isEqualTo(original.length);
+                assertThat(p.flushed()).isEqualTo(r.bytesProduced());
+                assertThat(p.activeWorkers()).isZero(); // single-threaded build
+            }
+        }
+    }
+
     @Nested
     class WithDictionary {