1package snapshotcodec
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 toolchain and level, so content addressing stays
16// stable between checkpoints of one build.
17
18// inflateInitialCap is the up-front allocation floor for one payload.
19const inflateInitialCap = 1 << 20
20
21// deflateMaxRatio bounds how many bytes one compressed byte can inflate to
22// (deflate tops out near 1032:1), so a payload the stream can really produce
23// is allocated once while a corrupt declared length stays bounded.
24const deflateMaxRatio = 1032
25
26var objectCompressorPool = sync.Pool{New: func() any {
27 w, err := zlib.NewWriterLevel(io.Discard, zlib.BestSpeed)
28 if err != nil {
29 panic(err) // BestSpeed is a valid level
30 }
31 return w
32}}
33
34var objectDecompressorPool sync.Pool
35
36// compressObjectPayload appends the zlib stream of payload to prefix.
37func compressObjectPayload(prefix, payload []byte) ([]byte, error) {
38 out := bytes.NewBuffer(prefix)
39 out.Grow(len(payload)/4 + 64)
40 w := objectCompressorPool.Get().(*zlib.Writer)
41 w.Reset(out)
42 defer func() {
43 w.Reset(io.Discard)
44 objectCompressorPool.Put(w)
45 }()
46 if _, err := w.Write(payload); err != nil {
47 return nil, err
48 }
49 if err := w.Close(); err != nil {
50 return nil, err
51 }
52 return out.Bytes(), nil
53}
54
55// inflateObjectPayload decodes exactly size bytes and requires a clean end of
56// stream (which also verifies the zlib checksum) with no trailing data.
57func inflateObjectPayload(compressed []byte, size uint32) ([]byte, error) {
58 invalid := func(err error) error { return fmt.Errorf("%w: object payload: %v", ErrInvalidData, err) }
59 source := bytes.NewReader(compressed)
60 r, err := objectDecompressor(source)
61 if err != nil {
62 return nil, invalid(err)
63 }
64 defer objectDecompressorPool.Put(r)
65 payload, err := readExact(r, int(size), max(inflateInitialCap, len(compressed)*deflateMaxRatio))
66 if err != nil {
67 return nil, invalid(err)
68 }
69 var extra [1]byte
70 if n, err := r.Read(extra[:]); n != 0 || err != io.EOF {
71 if err == nil || err == io.EOF {
72 err = errors.New("payload longer than declared")
73 }
74 return nil, invalid(err)
75 }
76 if source.Len() != 0 {
77 return nil, ErrTrailingBytes
78 }
79 return payload, nil
80}
81
82// readExact reads exactly size bytes. It doubles the buffer as data arrives,
83// starting at initial and capping every step at size, so a corrupt declared
84// length allocates at most about four times max(initial, what the stream
85// produces) and the returned payload has no slack.
86func readExact(r io.Reader, size, initial int) ([]byte, error) {
87 buf := make([]byte, min(size, initial))
88 for n := 0; ; {
89 m, err := io.ReadFull(r, buf[n:])
90 n += m
91 if err != nil {
92 if err == io.EOF {
93 err = io.ErrUnexpectedEOF
94 }
95 return nil, err
96 }
97 if n == size {
98 return buf, nil
99 }
100 grown := make([]byte, min(2*len(buf), size))
101 copy(grown, buf)
102 buf = grown
103 }
104}
105
106func objectDecompressor(source io.Reader) (io.ReadCloser, error) {
107 if pooled, ok := objectDecompressorPool.Get().(io.ReadCloser); ok {
108 if err := pooled.(zlib.Resetter).Reset(source, nil); err != nil {
109 return nil, err
110 }
111 return pooled, nil
112 }
113 return zlib.NewReader(source)
114}