Files
sow-tools/internal/nwsync/compressedbuf.go
T
archvillainetteandClaude Opus 5 59b233e3db fix(nwsync): stream emit so peak memory tracks the largest resource (#76)
Emit read the whole artifact into memory and erf.Read then allocated a
second full copy of every payload, so a hak cost about 5x its size in
RAM. The 7 GB runner was OOM-killed on any hak over ~1.4 GB, which
blocks the NWSync backfill and every release that rebuilds a large hak.

Emit now opens the artifact, hashes it by streaming for the key check,
parses only the header and resource table via erf.ReadIndex, and reads,
hashes, compresses and stores one payload at a time. erf.Read keeps its
old shape but returns payloads as subslices instead of fresh copies,
which removes the second copy for the other callers too.

The zstd encoder pool also held one window-sized history per CPU — about
200 MB of live heap on a 24-core runner. EncodeAll is single-threaded
per call, so concurrency 1 gives byte-identical blobs for far less
memory.

Peak heap is now flat at ~22 MB for both an 8 MB and a 64 MB hak, and a
regression test asserts it does not scale with artifact size.

Closes #76

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-31 13:06:29 +02:00

81 lines
2.8 KiB
Go

package nwsync
import (
"bytes"
"encoding/binary"
"fmt"
"github.com/klauspost/compress/zstd"
)
// NWCompressedBuffer framing, as upstream's neverwinter/compressedbuf.nim
// writes it for NWSync blobs. All fields are little-endian uint32:
//
// magic "NSYC", version 3, algorithm 2 (zstd), uncompressed size,
// zstd header version 1, dictionary 0, then the raw zstd frame.
const (
blobMagic = 0x4359534E // "NSYC" little-endian
blobVersion = 3
algorithmZstd = 2
zstdHeaderVer = 1
zstdDictionary = 0
blobHeaderBytes = 24
)
// EncodeAll/DecodeAll are single-threaded per call, so the default pool of one
// encoder per CPU only buys idle memory: each holds a window-sized history, so
// on a 24-core runner that is ~200 MB of live heap doing nothing. Concurrency 1
// produces byte-identical output.
var (
blobEncoder, _ = zstd.NewWriter(nil, zstd.WithEncoderConcurrency(1))
blobDecoder, _ = zstd.NewReader(nil, zstd.WithDecoderConcurrency(1))
)
// compressBlob wraps data in NWCompressedBuffer framing.
func compressBlob(data []byte) []byte {
var out bytes.Buffer
header := []uint32{blobMagic, blobVersion, algorithmZstd, uint32(len(data)), zstdHeaderVer, zstdDictionary}
for _, field := range header {
_ = binary.Write(&out, binary.LittleEndian, field)
}
out.Write(blobEncoder.EncodeAll(data, nil))
return out.Bytes()
}
// decompressBlob unwraps NWCompressedBuffer framing. It exists so a blob this
// package wrote — or one upstream wrote — can be compared by its uncompressed
// bytes, which is the only comparison that is meaningful across zstd
// implementations.
func decompressBlob(blob []byte) ([]byte, error) {
if len(blob) < blobHeaderBytes {
return nil, fmt.Errorf("blob too small: %d bytes", len(blob))
}
header := make([]uint32, 6)
if err := binary.Read(bytes.NewReader(blob[:blobHeaderBytes]), binary.LittleEndian, header); err != nil {
return nil, fmt.Errorf("decode blob header: %w", err)
}
switch {
case header[0] != blobMagic:
return nil, fmt.Errorf("invalid blob magic: %#x", header[0])
case header[1] != blobVersion:
return nil, fmt.Errorf("unsupported blob version: %d", header[1])
case header[2] != algorithmZstd:
return nil, fmt.Errorf("unsupported compression algorithm: %d", header[2])
case header[4] != zstdHeaderVer:
return nil, fmt.Errorf("unsupported zstd header version: %d", header[4])
case header[5] != zstdDictionary:
return nil, fmt.Errorf("zstd dictionaries are not supported")
}
if header[3] == 0 {
return nil, nil
}
data, err := blobDecoder.DecodeAll(blob[blobHeaderBytes:], nil)
if err != nil {
return nil, fmt.Errorf("decompress blob: %w", err)
}
if uint32(len(data)) != header[3] {
return nil, fmt.Errorf("blob size mismatch: header says %d, got %d", header[3], len(data))
}
return data, nil
}