Zstandard storage transformer

Stream transformer that compresses byte streams with Zstandard on the way into storage and decompresses them on the way back, with constant memory usage.

Overview

Zstandard produces a better compression ratio than gzip at a similar speed. The transformer compresses uploads through klauspost/compress/zstd, a pure-Go library with no CGO and no build tags, so it behaves the same in interpreted dev mode and in compiled builds. It streams through io.Pipe, so the source reader feeds the encoder and the consumer pulls the compressed result without buffering the whole object.

The transformer implements StreamTransformerPort directly. One RegisterTransformer call plugs it into the same storage transformer chain that the framework uses for its built-in crypto transformer, with no glue code. It composes with other transformers through an integer priority, so chaining compress-then-encrypt is declarative ordering, not a hand-written pipeline. Each Transform and Reverse call creates an independent encoder or decoder, so the transformer is safe for concurrent use, and errors propagate through the piped reader.

The four predefined levels are zstd.SpeedFastest (1), zstd.SpeedDefault (2), zstd.SpeedBetterCompression (3), and zstd.SpeedBestCompression (4), with SpeedDefault as the default. The Reverse path caps decompressed output to guard against decompression bombs on untrusted downloads. See Decompression cap.

Reach for gzip when third parties consume the objects directly and need universal format support. Reach for the crypto transformer when you also need at-rest encryption.

Configuration

import (
    "github.com/klauspost/compress/zstd"
    "piko.sh/piko/wdk/storage/storage_transformer_zstd"
)

transformer, err := storage_transformer_zstd.NewZstdTransformer(storage_transformer_zstd.Config{
    Name:                 "zstd",            // optional, default "zstd"
    Priority:             100,               // optional, default 100
    Level:                zstd.SpeedDefault, // optional, default SpeedDefault
    MaxDecompressedBytes: 256 * 1024 * 1024, // optional, default 256 MiB
})
if err != nil {
    return err
}

storage_transformer_zstd.DefaultConfig() returns the same defaults to start from and override one field.

Config fields:

Name. The transformer identifier in the registry. Defaults to zstd.
Priority. The position in the transform chain. Lower values run first on writes. Defaults to 100.
Level. The zstd compression level. A zero value maps to zstd.SpeedDefault.
MaxDecompressedBytes. The cap on bytes produced by Reverse. A zero value uses the 256 MiB default. A negative value disables the cap.

WithMaxDecompressedBytes(maxBytes int64) sets the same cap as a construction option:

transformer, err := storage_transformer_zstd.NewZstdTransformer(
    storage_transformer_zstd.DefaultConfig(),
    storage_transformer_zstd.WithMaxDecompressedBytes(64*1024*1024),
)

Bootstrap

Zstd registers against the storage service transformer registry, not through piko.With*. RegisterTransformer is the only supported wiring. There is no NewService option for transformers. Register after building the service:

if err := service.RegisterTransformer(ctx, transformer); err != nil {
    return err
}

Use a transformer on writes and reads

PutObject takes a provider name and a *storage.PutParams and returns only an error. Set TransformConfig.EnabledTransformers to the transformer names that apply to the call. The TransformerOptions map keys on the transformer name and overrides the level for that one operation:

params := &storage.PutParams{
    Key:    "report.json",
    Reader: r,
    TransformConfig: &storage.TransformConfig{
        EnabledTransformers: []string{"zstd"},
        TransformerOptions: map[string]any{
            "zstd": map[string]any{"level": int(zstd.SpeedBestCompression)},
        },
    },
}
if err := service.PutObject(ctx, providerName, params); err != nil {
    return err
}

GetObject mirrors the write path. Pass the same EnabledTransformers so the chain reverses the zstd step on the way back:

reader, err := service.GetObject(ctx, providerName, storage.GetParams{
    Key: "report.json",
    TransformConfig: &storage.TransformConfig{
        EnabledTransformers: []string{"zstd"},
    },
})
if err != nil {
    return err
}
defer func() { _ = reader.Close() }()

Ordering with encryption

The chain sorts transformers by ascending priority and applies them in that order on writes, so a lower priority runs first and a higher priority runs later. Reads apply the chain in reverse. To compress first and then encrypt the smaller output, register the crypto transformer at a higher priority than zstd.

The bootstrap registers the framework crypto transformer at priority 100, the same as the zstd default. Equal priorities give no deterministic compress-then-encrypt order. To guarantee that compression runs first, register the crypto transformer yourself at a priority above your zstd priority, or lower the zstd priority below the crypto priority.

Decompression cap

Reverse wraps the zstd decoder so reads beyond MaxDecompressedBytes surface ErrDecompressedTooLarge instead of inflating without bound. The default cap is 256 MiB. This protects callers that buffer a download against a small upload that expands to gigabytes. Use errors.Is(err, storage_transformer_zstd.ErrDecompressedTooLarge) to distinguish the cap from a normal end of stream. Lower the cap for stricter limits on untrusted objects, or set a negative value to disable it for fully trusted input.