1package snapshot
2
3import (
4 "bytes"
5 "compress/zlib"
6 "errors"
7 "fmt"
8 "io"
9 "sync"
10)
11
12// Object payloads are stored zlib-compressed. History cells are highly
13// repetitive, so a sealed 10k-row history shrinks several times on disk.
14// BestSpeed keeps snapshot encoding cheap; Go's deflate output is
15// deterministic for one level, so content addressing stays stable.
16
17var objectCompressorPool = sync.Pool{New: func() any {
18 w, err := zlib.NewWriterLevel(io.Discard, zlib.BestSpeed)
19 if err != nil {
20 panic(err) // BestSpeed is a valid level
21 }
22 return w
23}}
24
25var objectDecompressorPool sync.Pool
26
27// compressObjectPayload appends the zlib stream of payload to prefix.
28func compressObjectPayload(prefix, payload []byte) ([]byte, error) {
29 out := bytes.NewBuffer(prefix)
30 out.Grow(len(payload)/4 + 64)
31 w := objectCompressorPool.Get().(*zlib.Writer)
32 w.Reset(out)
33 defer func() {
34 w.Reset(io.Discard)
35 objectCompressorPool.Put(w)
36 }()
37 if _, err := w.Write(payload); err != nil {
38 return nil, err
39 }
40 if err := w.Close(); err != nil {
41 return nil, err
42 }
43 return out.Bytes(), nil
44}
45
46// inflateObjectPayload decodes exactly size bytes and requires a clean end of
47// stream (which also verifies the zlib checksum) with no trailing data.
48func inflateObjectPayload(compressed []byte, size uint32) ([]byte, error) {
49 invalid := func(err error) error { return fmt.Errorf("%w: object payload: %v", ErrInvalidData, err) }
50 source := bytes.NewReader(compressed)
51 r, err := objectDecompressor(source)
52 if err != nil {
53 return nil, invalid(err)
54 }
55 defer objectDecompressorPool.Put(r)
56 payload := make([]byte, size)
57 if _, err := io.ReadFull(r, payload); err != nil {
58 return nil, invalid(err)
59 }
60 var extra [1]byte
61 if n, err := r.Read(extra[:]); n != 0 || err != io.EOF {
62 if err == nil || err == io.EOF {
63 err = errors.New("payload longer than declared")
64 }
65 return nil, invalid(err)
66 }
67 if source.Len() != 0 {
68 return nil, ErrTrailingBytes
69 }
70 return payload, nil
71}
72
73func objectDecompressor(source io.Reader) (io.ReadCloser, error) {
74 if pooled, ok := objectDecompressorPool.Get().(io.ReadCloser); ok {
75 if err := pooled.(zlib.Resetter).Reset(source, nil); err != nil {
76 return nil, err
77 }
78 return pooled, nil
79 }
80 return zlib.NewReader(source)
81}