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 bounds the up-front allocation for one payload, so a
19// corrupt declared length cannot allocate more than the stream produces.
20const inflateInitialCap = 1 << 20
21
22var objectCompressorPool = sync.Pool{New: func() any {
23 w, err := zlib.NewWriterLevel(io.Discard, zlib.BestSpeed)
24 if err != nil {
25 panic(err) // BestSpeed is a valid level
26 }
27 return w
28}}
29
30var objectDecompressorPool sync.Pool
31
32// compressObjectPayload appends the zlib stream of payload to prefix.
33func compressObjectPayload(prefix, payload []byte) ([]byte, error) {
34 out := bytes.NewBuffer(prefix)
35 out.Grow(len(payload)/4 + 64)
36 w := objectCompressorPool.Get().(*zlib.Writer)
37 w.Reset(out)
38 defer func() {
39 w.Reset(io.Discard)
40 objectCompressorPool.Put(w)
41 }()
42 if _, err := w.Write(payload); err != nil {
43 return nil, err
44 }
45 if err := w.Close(); err != nil {
46 return nil, err
47 }
48 return out.Bytes(), nil
49}
50
51// inflateObjectPayload decodes exactly size bytes and requires a clean end of
52// stream (which also verifies the zlib checksum) with no trailing data.
53func inflateObjectPayload(compressed []byte, size uint32) ([]byte, error) {
54 invalid := func(err error) error { return fmt.Errorf("%w: object payload: %v", ErrInvalidData, err) }
55 source := bytes.NewReader(compressed)
56 r, err := objectDecompressor(source)
57 if err != nil {
58 return nil, invalid(err)
59 }
60 defer objectDecompressorPool.Put(r)
61 out := bytes.NewBuffer(make([]byte, 0, min(int(size), inflateInitialCap)))
62 if _, err := io.Copy(out, io.LimitReader(r, int64(size))); err != nil {
63 return nil, invalid(err)
64 }
65 if out.Len() != int(size) {
66 return nil, invalid(io.ErrUnexpectedEOF)
67 }
68 payload := out.Bytes()
69 if cap(payload) > len(payload)+len(payload)/4 {
70 // Restored payloads are retained; drop the growth slack.
71 payload = bytes.Clone(payload)
72 }
73 var extra [1]byte
74 if n, err := r.Read(extra[:]); n != 0 || err != io.EOF {
75 if err == nil || err == io.EOF {
76 err = errors.New("payload longer than declared")
77 }
78 return nil, invalid(err)
79 }
80 if source.Len() != 0 {
81 return nil, ErrTrailingBytes
82 }
83 return payload, nil
84}
85
86func objectDecompressor(source io.Reader) (io.ReadCloser, error) {
87 if pooled, ok := objectDecompressorPool.Get().(io.ReadCloser); ok {
88 if err := pooled.(zlib.Resetter).Reset(source, nil); err != nil {
89 return nil, err
90 }
91 return pooled, nil
92 }
93 return zlib.NewReader(source)
94}