Working with zstd.Decompress has some limitations that aren't obvious

[mrjbq7]

Zig Version

0.15.0-dev.1267+da408bd6f

Steps to Reproduce and Observed Behavior

Using a zstd.Decompress with an 80 mb buffer:

    const args = try std.process.argsAlloc(allocator);
    defer std.process.argsFree(allocator, args);

    if (args.len < 2) {
        std.debug.print("Usage: {s} <file>\n", .{args[0]});
        return;
    }

    const path = args[1];

    const file = if (std.fs.path.isAbsolute(path))
        try std.fs.openFileAbsolute(path, .{})
    else
        try std.fs.cwd().openFile(path, .{});

    const stat = try file.stat();
    std.debug.print("file is {B}\n", .{stat.size});

    var file_buffer: [4096]u8 = undefined;
    var file_reader = file.reader(&file_buffer);

    const zstd_buffer = try allocator.alloc(u8, 10 * (zstd.default_window_len + zstd.block_size_max));
    defer allocator.free(zstd_buffer);
    var zstd_reader: std.compress.zstd.Decompress = .init(&file_reader.interface, zstd_buffer, .{});

these don't work, and i think should:

1) Reading byte-by-byte

    for (0..1_000_000) |i| {
        const b = try zstd_reader.reader.takeByte();
        if ((i % 1000) == 0) {
            std.debug.print("{d}\n", .{b});
        }
    }

has this error:

error: ReadFailed
zig/build/stage3/lib/zig/std/math.zig:578:21: 0x104d2f63f in sub__anon_11112 (repro)
    if (ov[1] != 0) return error.Overflow;
                    ^
zig/build/stage3/lib/zig/std/compress/zstd/Decompress.zig:671:21: 0x104d975d7 in decodeSequence (repro)
                    return error.MalformedSequence;
                    ^
zig/build/stage3/lib/zig/std/compress/zstd/Decompress.zig:219:42: 0x104d98d5f in readInFrame (repro)
                        bytes_written += try decode.decodeSequence(w.buffer, write_pos + bytes_written, &bit_stream);
                                         ^
zig/build/stage3/lib/zig/std/compress/zstd/Decompress.zig:138:21: 0x104d921fb in stream (repro)
                    return error.ReadFailed;
                    ^
zig/build/stage3/lib/zig/std/Io/Reader.zig:1063:25: 0x104daa55f in fillUnbuffered (repro)
            else => |e| return e,
                        ^
zig/build/stage3/lib/zig/std/Io/Reader.zig:1046:5: 0x104d9d267 in fill (repro)
    return fillUnbuffered(r, n);
    ^
zig/build/stage3/lib/zig/std/Io/Reader.zig:1111:5: 0x104d928bf in peekByte (repro)
    try fill(r, 1);
    ^
zig/build/stage3/lib/zig/std/Io/Reader.zig:1119:20: 0x104d8e95f in takeByte (repro)
    const result = try peekByte(r);
                   ^
repro.zig:42:19: 0x104d8e183 in main (repro)
        const b = try zstd_reader.reader.takeByte();

2) Reading using slices

    var buffer: [1024]u8 = undefined;

    for (0..1024) |i| {
        std.debug.print("{d}\n", .{i});
        try zstd_reader.reader.readSliceAll(&buffer);
    }

fails with

error: ReadFailed
zig/std/math.zig:578:21: 0x10091b63f in sub__anon_11112 (repro)
    if (ov[1] != 0) return error.Overflow;
                    ^
zig/std/compress/zstd/Decompress.zig:671:21: 0x100983a87 in decodeSequence (repro)
                    return error.MalformedSequence;
                    ^
zig/std/compress/zstd/Decompress.zig:219:42: 0x10098520f in readInFrame (repro)
                        bytes_written += try decode.decodeSequence(w.buffer, write_pos + bytes_written, &bit_stream);
                                         ^
zig/std/compress/zstd/Decompress.zig:138:21: 0x10097e1cf in stream (repro)
                    return error.ReadFailed;
                    ^
zig/std/Io/Reader.zig:408:21: 0x1009898db in readVecInner (repro)
        else => |e| return e,
                    ^
zig/std/Io/Reader.zig:662:33: 0x10097edff in readSliceShort (repro)
            error.ReadFailed => return error.ReadFailed,
                                ^
zig/std/Io/Reader.zig:630:15: 0x10097aa37 in readSliceAll (repro)
    const n = try readSliceShort(r, buffer);
              ^
repro.zig:45:9: 0x10097a1a7 in main (repro)
        try zstd_reader.reader.readSliceAll(&buffer);
        ^

3) streamExact 1k to stdout using a 80mb stdout buffer

    const stdout_buffer = try allocator.alloc(u8, 10 * (zstd.default_window_len + zstd.block_size_max));
    defer allocator.free(stdout_buffer);
    var stdout = std.fs.File.stdout().writer(stdout_buffer);
    try zstd_reader.reader.streamExact(&stdout.interface, 1024);

fails with

error: ReadFailed
zig/std/compress/zstd/Decompress.zig:182:43: 0x10252414b in readInFrame (repro)
    if (@intFromEnum(limit) < block_size) return error.OutputBufferUndersize;
                                          ^
zig/std/compress/zstd/Decompress.zig:128:21: 0x10251e0bf in stream (repro)
                    return error.ReadFailed;
                    ^
zig/std/Io/Reader.zig:159:15: 0x10249ac17 in stream (repro)
    const n = try r.vtable.stream(r, w, limit);
              ^
zig/std/Io/Reader.zig:196:41: 0x10251ad5f in streamExact (repro)
    while (remaining != 0) remaining -= try r.stream(w, .limited(remaining));
                                        ^
repro.zig:57:5: 0x10251a1fb in main (repro)
    try zstd_reader.reader.streamExact(&stdout.interface, 1024);
    ^

4) using allocRemaining to read one block size:

    const buf = try zstd_reader.reader.allocRemaining(allocator, .limited(zstd.block_size_max));
    std.debug.print("took {d}\n", .{buf.len});

fails with

error: ReadFailed
zig/std/compress/zstd/Decompress.zig:817:13: 0x1023269d3 in init (repro)
            return error.WindowOversize
            ^
zig/std/compress/zstd/Decompress.zig:161:26: 0x102326143 in initFrame (repro)
                .frame = try Frame.init(header, window_size_max, d.verify_checksum),
                         ^
zig/std/compress/zstd/Decompress.zig:121:17: 0x102321f13 in stream (repro)
                return error.ReadFailed;
                ^
zig/std/Io/Reader.zig:327:37: 0x102322ef7 in appendRemaining__anon_19657 (repro)
                error.ReadFailed => return error.ReadFailed,
                                    ^
zig/std/Io/Reader.zig:275:5: 0x10231e967 in allocRemaining (repro)
    try appendRemaining(r, gpa, null, &buffer, limit);
    ^
repro.zig:41:17: 0x10231e0ef in main (repro)
    const buf = try zstd_reader.reader.allocRemaining(allocator, .limited(zstd.block_size_max));
                ^

5) using take() to get larger than block_size_max

    const buf = try zstd_reader.reader.take(2 * zstd.block_size_max);
    std.debug.print("took {d}\n", .{buf.len});

fails with the same error as above:

error: ReadFailed
zig/std/math.zig:578:21: 0x100f2f63f in sub__anon_11112 (repro)
    if (ov[1] != 0) return error.Overflow;
                    ^
zig/std/compress/zstd/Decompress.zig:671:21: 0x100f9754f in decodeSequence (repro)
                    return error.MalformedSequence;
                    ^
zig/std/compress/zstd/Decompress.zig:219:42: 0x100f98cd7 in readInFrame (repro)
                        bytes_written += try decode.decodeSequence(w.buffer, write_pos + bytes_written, &bit_stream);
                                         ^
zig/std/compress/zstd/Decompress.zig:138:21: 0x100f921a3 in stream (repro)
                    return error.ReadFailed;
                    ^
zig/std/Io/Reader.zig:1063:25: 0x100faa3df in fillUnbuffered (repro)
            else => |e| return e,
                        ^
zig/std/Io/Reader.zig:1046:5: 0x100f9d2ef in fill (repro)
    return fillUnbuffered(r, n);
    ^
zig/std/Io/Reader.zig:478:5: 0x100f92723 in peek (repro)
    try r.fill(n);
    ^
zig/std/Io/Reader.zig:527:20: 0x100f8e907 in take (repro)
    const result = try r.peek(n);
                   ^
repro.zig:44:17: 0x100f8e0d7 in main (repro)
    const buf = try zstd_reader.reader.take(2 * zstd.block_size_max);
                ^

There are other test cases that we could run that similarly fail. I guess, after looking at it, I would suggest that all of these are code that a user would think should work, and that the main issue appears to be mixing the concept of the zstd.Decompress buffer that maybe should be an interim decompressed bytes location that can be take() or readSliceAll() from, but then when an output writer is used, it might try and use the output writer for that decompressed data which also means that you can't have smaller (or larger) than block size writes...

Expected Behavior

I think these cases should work.

[mrjbq7]

6) Here's some kind of infinite loop (panic):

const std = @import("std");
const zstd = std.compress.zstd;

pub fn main() anyerror!void {
    const input_file = try std.fs.cwd().openFile("foo.zst", .{});
    var input_buffer: [4096]u8 = undefined;
    var input_reader = input_file.reader(&input_buffer);

    var buffer: [zstd.default_window_len + zstd.block_size_max]u8 = undefined;
    var decompress: zstd.Decompress = .init(&input_reader.interface, &buffer, .{});

    const header = try decompress.reader.take(3);
    if (!std.mem.eql(u8, header, "DBN")) return error.BadMagic;

    const version = try decompress.reader.takeByte();
    if (version != 1) return error.BadVersion;

    const n = try decompress.reader.takeInt(u32, .little);
    if (n < 100) return error.BadLength;

    const discarded = try decompress.reader.discard(.limited(n));
    std.debug.assert(discarded == n);

    while (true) {
        const words = try decompress.reader.peekByte();
        const length = @as(usize, words) * 4;
        const record = try decompress.reader.take(length);
        std.debug.assert(record.len == length);
        std.debug.print("read {d} bytes\n", .{record.len});
    }
}

Download this 5mb zstd file, it's chopped from a larger file, so once the infinite loop is fixed, it probably breaks due to EndOfStream.

https://downloads.factorcode.org/misc/foo.zst

The backtrace:

Process 23363 stopped
* thread #18, queue = 'com.apple.main-thread', stop reason = signal SIGSTOP
    frame #0: 0x000000010000955c repro2`Io.Writer.fixedDrain(w=0x000000016f5d96c8, data=[][]const u8 @ 0x000000016f598308, splat=1) at Writer.zig:2267:26
   2264         if (bytes.len > dest.len) return error.WriteFailed;
   2265     }
   2266     const pattern = data[data.len - 1];
-> 2267     const dest = w.buffer[w.end..];
   2268     switch (pattern.len) {
   2269         0 => return 0,
   2270         1 => {
Target 0: (repro2) stopped.

although i get different backtraces depending on when i ctrl-c

* thread #18, queue = 'com.apple.main-thread', stop reason = signal SIGSTOP
    frame #0: 0x00000001000a28e8 repro2`Io.Writer.writableSliceGreedyPreserve(w=0x000000016f5d96c8, preserve_len=8388608, minimum_length=65576) at Writer.zig:323:25
   320  /// If `preserve_len` is zero, this is equivalent to `writableSliceGreedy`.
   321  pub fn writableSliceGreedyPreserve(w: *Writer, preserve_len: usize, minimum_length: usize) Error![]u8 {
   322      assert(w.buffer.len >= preserve_len + minimum_length);
-> 323      while (w.buffer.len - w.end < minimum_length) {
   324          try drainPreserve(w, preserve_len);
   325      } else {
   326          @branchHint(.likely);
Target 0: (repro2) stopped.
(lldb) bt
* thread #18, queue = 'com.apple.main-thread', stop reason = signal SIGSTOP
  * frame #0: 0x00000001000a28e8 repro2`Io.Writer.writableSliceGreedyPreserve(w=0x000000016f5d96c8, preserve_len=8388608, minimum_length=65576) at Writer.zig:323:25
    frame #1: 0x000000010009f0dc repro2`compress.zstd.Decompress.readInFrame(d=0x000000016fdfaa18, w=0x000000016f5d96c8, limit=60170, state=0x000000016fdfaa48) at Decompress.zig:260:68
    frame #2: 0x000000010009a760 repro2`compress.zstd.Decompress.stream(r=0x000000016fdfaa20, w=0x000000016f5d96c8, limit=60170) at Decompress.zig:167:31
    frame #3: 0x00000001000a3a98 repro2`Io.Reader.fillUnbuffered(r=0x000000016fdfaa20, n=1) at Reader.zig:1068:33
    frame #4: 0x000000010009b860 repro2`Io.Reader.fill(r=0x000000016fdfaa20, n=1) at Reader.zig:1049:26
    frame #5: 0x000000010009a014 repro2`Io.Reader.peekByte(r=0x000000016fdfaa20) at Reader.zig:1105:13
    frame #6: 0x00000001000997a0 repro2`repro2.main at repro2.zig:24:53
    frame #7: 0x0000000100099104 repro2`start.main [inlined] start.callMain at start.zig:627:37
    frame #8: 0x00000001000990f8 repro2`start.main [inlined] start.callMainWithArgs at start.zig:587:20
    frame #9: 0x000000010009907c repro2`start.main(c_argc=1, c_argv=0x000000016fdfe6d0, c_envp=0x000000016fdfe6e0) at start.zig:602:28
    frame #10: 0x000000019ac72b98 dyld`start + 6076

[mrjbq7]

the infinite loop is changed with #24706, it now is a panic:

...
read 56 bytes
read 56 bytes
read 56 bytes
read 56 bytes
read 56 bytes
read 56 bytes
read 56 bytes
read 56 bytes
read 56 bytes
thread 17748576 panic: reached unreachable code
zig/build/stage3/lib/zig/std/debug.zig:559:14: 0x1044c7e07 in assert (repro2)
    if (!ok) unreachable; // assertion failure
             ^
zig/build/stage3/lib/zig/std/compress/zstd/Decompress.zig:103:11: 0x10455b20f in rebase (repro2)
    assert(capacity <= r.buffer.len - d.window_len);
          ^
zig/build/stage3/lib/zig/std/Io/Reader.zig:1298:27: 0x1045639a3 in rebase (repro2)
    return r.vtable.rebase(r, capacity);
                          ^
zig/build/stage3/lib/zig/std/compress/zstd/Decompress.zig:142:13: 0x10455afef in readVec (repro2)
    r.rebase(d.window_len) catch unreachable;
            ^
zig/build/stage3/lib/zig/std/Io/Reader.zig:1070:56: 0x104563b47 in fillUnbuffered (repro2)
    while (r.end < r.seek + n) _ = try r.vtable.readVec(r, &bufs);
                                                       ^
zig/build/stage3/lib/zig/std/Io/Reader.zig:1056:26: 0x10455b977 in fill (repro2)
    return fillUnbuffered(r, n);
                         ^
zig/build/stage3/lib/zig/std/Io/Reader.zig:1096:13: 0x10455a12b in peekByte (repro2)
    try fill(r, 1);
            ^
repro2.zig:24:53: 0x1045598b7 in main (repro2)
        const words = try decompress.reader.peekByte();
                                                    ^
zig/build/stage3/lib/zig/std/start.zig:627:37: 0x10455921b in main (repro2)
            const result = root.main() catch |err| {
                                    ^
???:?:?: 0x19ac72b97 in ??? (???)
???:?:?: 0x0 in ??? (???)
[1]    65591 abort      zig/build/stage3/bin/zig run repro2.zig