1// SPDX-License-Identifier: MIT
2
3package kv
4
5import (
6 "bufio"
7 "bytes"
8 "encoding/binary"
9 "errors"
10 "fmt"
11 "hash/crc32"
12 "io"
13)
14
15// Operation kinds.
16const (
17 opPut byte = 1
18 opDelete byte = 2
19)
20
21// Format limits.
22const (
23 maxRecord = 64 << 20 // maximum record payload size
24 maxKeyLen = 64 << 10 // maximum key length
25)
26
27const (
28 logMagic = "EKVLOG01"
29 snapshotMagic = "EKVSNP01"
30 logHeaderSize = 8 + 8 + 4
31 recordFrame = 4 + 4 // len + crc
32 recordHead = 8 + 4 // seq + batch count
33 batchHeader = 4 // op count
34 recordMinBody = 16 // smallest readable payload length
35 snapHeaderSize = 8 + 8 + 8 + 4
36 minOpSize = 3 // kind + klen + 1 key byte
37)
38
39var castagnoli = crc32.MakeTable(crc32.Castagnoli)
40
41// errUnreadable marks a log record that is not readable (see decodeRecord).
42var errUnreadable = errors.New("unreadable record")
43
44// op is one mutation of a batch. value is nil for opDelete.
45type op struct {
46 kind byte
47 key, value []byte
48}
49
50// opSize returns the encoded size of o.
51func opSize(o op) int {
52 n := 1 + uvarintLen(uint64(len(o.key))) + len(o.key)
53 if o.kind == opPut {
54 n += uvarintLen(uint64(len(o.value))) + len(o.value)
55 }
56 return n
57}
58
59func uvarintLen(v uint64) int {
60 n := 1
61 for ; v >= 0x80; v >>= 7 {
62 n++
63 }
64 return n
65}
66
67// appendLogHeader appends the 20-byte log header for base to dst.
68func appendLogHeader(dst []byte, base uint64) []byte {
69 start := len(dst)
70 dst = append(dst, logMagic...)
71 dst = binary.LittleEndian.AppendUint64(dst, base)
72 return binary.LittleEndian.AppendUint32(dst, crc32.Checksum(dst[start:], castagnoli))
73}
74
75// decodeLogHeader validates the log header at the start of b and returns its
76// base. Errors wrap ErrCorrupt.
77func decodeLogHeader(b []byte) (base uint64, err error) {
78 if len(b) < logHeaderSize {
79 return 0, fmt.Errorf("%w: short log header", ErrCorrupt)
80 }
81 if string(b[:8]) != logMagic {
82 return 0, fmt.Errorf("%w: bad log magic", ErrCorrupt)
83 }
84 if crc32.Checksum(b[:16], castagnoli) != binary.LittleEndian.Uint32(b[16:20]) {
85 return 0, fmt.Errorf("%w: bad log header checksum", ErrCorrupt)
86 }
87 return binary.LittleEndian.Uint64(b[8:16]), nil
88}
89
90// beginRecord starts a group record: it appends the frame placeholder, the
91// seq of the first batch and the batch count placeholder to dst. Batches are
92// added with appendBatch and the record is completed by finishRecord, which
93// expects the record to start at index 0 of the slice.
94func beginRecord(dst []byte, seq uint64) []byte {
95 dst = append(dst, make([]byte, recordFrame)...)
96 dst = binary.LittleEndian.AppendUint64(dst, seq)
97 return binary.LittleEndian.AppendUint32(dst, 0)
98}
99
100// appendBatch appends one batch (count, ops) to the record being built. ops
101// must be non-empty with valid keys; the caller enforces this (batches are
102// validated before encoding).
103func appendBatch(dst []byte, ops []op) []byte {
104 dst = binary.LittleEndian.AppendUint32(dst, uint32(len(ops)))
105 for _, o := range ops {
106 dst = append(dst, o.kind)
107 dst = binary.AppendUvarint(dst, uint64(len(o.key)))
108 dst = append(dst, o.key...)
109 if o.kind == opPut {
110 dst = binary.AppendUvarint(dst, uint64(len(o.value)))
111 dst = append(dst, o.value...)
112 }
113 }
114 return dst
115}
116
117// finishRecord patches the batch count, length and checksum into rec, which
118// holds exactly one record started by beginRecord.
119func finishRecord(rec []byte, batches int) {
120 binary.LittleEndian.PutUint32(rec[recordFrame+8:], uint32(batches))
121 payload := rec[recordFrame:]
122 binary.LittleEndian.PutUint32(rec, uint32(len(payload)))
123 binary.LittleEndian.PutUint32(rec[4:], crc32.Checksum(payload, castagnoli))
124}
125
126// appendRecord appends one encoded group record to dst. Batch i has seq
127// seq+i. The payload must not exceed maxRecord and every batch must hold
128// valid ops.
129func appendRecord(dst []byte, seq uint64, batches ...[]op) []byte {
130 rec := beginRecord(nil, seq)
131 for _, ops := range batches {
132 rec = appendBatch(rec, ops)
133 }
134 finishRecord(rec, len(batches))
135 return append(dst, rec...)
136}
137
138// peekRecord returns the total encoded size (frame + payload) announced by
139// the record frame at the start of b. ok is false if fewer than 8 bytes are
140// present or the announced payload length is outside [16, maxRecord]. It does
141// not check that the payload is present.
142func peekRecord(b []byte) (total int, ok bool) {
143 if len(b) < recordFrame {
144 return 0, false
145 }
146 n := binary.LittleEndian.Uint32(b)
147 if n < recordMinBody || n > maxRecord {
148 return 0, false
149 }
150 return recordFrame + int(n), true
151}
152
153// decodeRecord decodes the record at the start of b. A record is readable
154// only if the frame is present, 16 ≤ len ≤ maxRecord, len payload bytes are
155// present, the CRC matches, and the payload holds batches ≥ 1 batches of
156// count ≥ 1 valid ops each, consuming exactly len bytes. Otherwise the error
157// wraps errUnreadable.
158//
159// Batch i has seq seq+i. The decoded keys and values alias b. n is the number
160// of bytes consumed.
161func decodeRecord(b []byte) (seq uint64, batches [][]op, n int, err error) {
162 total, ok := peekRecord(b)
163 if !ok {
164 return 0, nil, 0, fmt.Errorf("%w: bad or short frame", errUnreadable)
165 }
166 if len(b) < total {
167 return 0, nil, 0, fmt.Errorf("%w: short payload", errUnreadable)
168 }
169 payload := b[recordFrame:total]
170 if crc32.Checksum(payload, castagnoli) != binary.LittleEndian.Uint32(b[4:8]) {
171 return 0, nil, 0, fmt.Errorf("%w: checksum mismatch", errUnreadable)
172 }
173 seq, batches, err = decodePayload(payload)
174 if err != nil {
175 return 0, nil, 0, err
176 }
177 return seq, batches, total, nil
178}
179
180// decodePayload decodes seq, batch count and batches from a whole payload.
181func decodePayload(p []byte) (uint64, [][]op, error) {
182 seq := binary.LittleEndian.Uint64(p)
183 nb := binary.LittleEndian.Uint32(p[8:])
184 p = p[recordHead:]
185 if nb == 0 || uint64(nb) > uint64(len(p)/(batchHeader+minOpSize)) {
186 return 0, nil, fmt.Errorf("%w: bad batch count %d", errUnreadable, nb)
187 }
188 batches := make([][]op, 0, nb)
189 for i := range nb {
190 if len(p) < batchHeader {
191 return 0, nil, fmt.Errorf("%w: truncated batch %d", errUnreadable, i)
192 }
193 count := binary.LittleEndian.Uint32(p)
194 p = p[batchHeader:]
195 if count == 0 || uint64(count) > uint64(len(p)/minOpSize) {
196 return 0, nil, fmt.Errorf("%w: bad op count %d in batch %d", errUnreadable, count, i)
197 }
198 ops := make([]op, 0, count)
199 for range count {
200 if len(p) == 0 {
201 return 0, nil, fmt.Errorf("%w: truncated op", errUnreadable)
202 }
203 o := op{kind: p[0]}
204 if o.kind != opPut && o.kind != opDelete {
205 return 0, nil, fmt.Errorf("%w: bad op kind %d", errUnreadable, o.kind)
206 }
207 var ok bool
208 if o.key, p, ok = cutBytes(p[1:], maxKeyLen); !ok || len(o.key) == 0 {
209 return 0, nil, fmt.Errorf("%w: bad key", errUnreadable)
210 }
211 if o.kind == opPut {
212 if o.value, p, ok = cutBytes(p, maxRecord); !ok {
213 return 0, nil, fmt.Errorf("%w: bad value", errUnreadable)
214 }
215 }
216 ops = append(ops, o)
217 }
218 batches = append(batches, ops)
219 }
220 if len(p) != 0 {
221 return 0, nil, fmt.Errorf("%w: %d trailing payload bytes", errUnreadable, len(p))
222 }
223 return seq, batches, nil
224}
225
226// cutBytes reads a uvarint length (at most limit) and that many bytes from
227// the start of b. rest is what follows. The returned slice has no spare
228// capacity.
229func cutBytes(b []byte, limit int) (v, rest []byte, ok bool) {
230 l, n := binary.Uvarint(b)
231 if n <= 0 || l > uint64(limit) || l > uint64(len(b)-n) {
232 return nil, nil, false
233 }
234 end := n + int(l)
235 return b[n:end:end], b[end:], true
236}
237
238// ownOps returns a copy of ops whose keys and values live in one new
239// allocation, so the result does not pin the buffer the ops were decoded from.
240func ownOps(ops []op) []op {
241 size := 0
242 for _, o := range ops {
243 size += len(o.key) + len(o.value)
244 }
245 buf := make([]byte, 0, size)
246 out := make([]op, len(ops))
247 for i, o := range ops {
248 out[i].kind = o.kind
249 buf = append(buf, o.key...)
250 out[i].key = buf[len(buf)-len(o.key) : len(buf) : len(buf)]
251 if o.kind == opPut {
252 buf = append(buf, o.value...)
253 out[i].value = buf[len(buf)-len(o.value) : len(buf) : len(buf)]
254 }
255 }
256 return out
257}
258
259// entryIter yields snapshot entries in strictly ascending key order. It stops
260// when yield returns false. The slices passed to yield need only stay valid
261// until yield returns.
262type entryIter = func(yield func(key, value []byte, version uint64) bool)
263
264// writeSnapshot writes a snapshot file of count entries at seq to w. entries
265// must yield exactly count entries sorted strictly ascending by key, with
266// 1 ≤ len(key) ≤ maxKeyLen and version ≤ seq; otherwise an error is returned
267// (the output is then incomplete and must be discarded).
268func writeSnapshot(w io.Writer, seq, count uint64, entries entryIter) error {
269 hdr := append(make([]byte, 0, snapHeaderSize), snapshotMagic...)
270 hdr = binary.LittleEndian.AppendUint64(hdr, seq)
271 hdr = binary.LittleEndian.AppendUint64(hdr, count)
272 hdr = binary.LittleEndian.AppendUint32(hdr, crc32.Checksum(hdr, castagnoli))
273 bw := bufio.NewWriterSize(w, 64<<10)
274 if _, err := bw.Write(hdr); err != nil {
275 return err
276 }
277
278 var (
279 sum uint32
280 buf []byte
281 prev []byte
282 written uint64
283 err error
284 )
285 entries(func(key, value []byte, version uint64) bool {
286 switch {
287 case written == count:
288 err = errors.New("snapshot: more entries than count")
289 case len(key) == 0 || len(key) > maxKeyLen:
290 err = fmt.Errorf("snapshot: invalid key length %d", len(key))
291 case written > 0 && bytes.Compare(prev, key) >= 0:
292 err = errors.New("snapshot: keys not strictly ascending")
293 case version > seq:
294 err = fmt.Errorf("snapshot: version %d above seq %d", version, seq)
295 }
296 if err != nil {
297 return false
298 }
299 buf = binary.AppendUvarint(buf[:0], uint64(len(key)))
300 buf = append(buf, key...)
301 buf = binary.AppendUvarint(buf, version)
302 buf = binary.AppendUvarint(buf, uint64(len(value)))
303 buf = append(buf, value...)
304 sum = crc32.Update(sum, castagnoli, buf)
305 if _, err = bw.Write(buf); err != nil {
306 return false
307 }
308 prev = append(prev[:0], key...)
309 written++
310 return true
311 })
312 if err != nil {
313 return err
314 }
315 if written != count {
316 return fmt.Errorf("snapshot: %d entries yielded, want %d", written, count)
317 }
318 if _, err := bw.Write(binary.LittleEndian.AppendUint32(nil, sum)); err != nil {
319 return err
320 }
321 return bw.Flush()
322}
323
324// snapshotView is a fully validated snapshot file held in memory.
325type snapshotView struct {
326 seq, count uint64
327 body []byte // entries, without header and trailer
328}
329
330// openSnapshot validates the snapshot file contents b: header and trailer
331// checksums, exactly count entries, strictly ascending non-empty keys,
332// versions ≤ seq and no trailing bytes. Errors wrap ErrCorrupt.
333func openSnapshot(b []byte) (snapshotView, error) {
334 if len(b) < snapHeaderSize+4 {
335 return snapshotView{}, fmt.Errorf("%w: short snapshot", ErrCorrupt)
336 }
337 if string(b[:8]) != snapshotMagic {
338 return snapshotView{}, fmt.Errorf("%w: bad snapshot magic", ErrCorrupt)
339 }
340 if crc32.Checksum(b[:snapHeaderSize-4], castagnoli) != binary.LittleEndian.Uint32(b[snapHeaderSize-4:]) {
341 return snapshotView{}, fmt.Errorf("%w: bad snapshot header checksum", ErrCorrupt)
342 }
343 v := snapshotView{
344 seq: binary.LittleEndian.Uint64(b[8:]),
345 count: binary.LittleEndian.Uint64(b[16:]),
346 body: b[snapHeaderSize : len(b)-4],
347 }
348 if crc32.Checksum(v.body, castagnoli) != binary.LittleEndian.Uint32(b[len(b)-4:]) {
349 return snapshotView{}, fmt.Errorf("%w: bad snapshot checksum", ErrCorrupt)
350 }
351 if err := v.walk(nil); err != nil {
352 return snapshotView{}, err
353 }
354 return v, nil
355}
356
357// each calls fn for every entry in ascending key order. The slices alias the
358// file contents: fn must not modify them and must copy them to retain them
359// independently of the file buffer.
360func (v snapshotView) each(fn func(key, value []byte, version uint64)) {
361 _ = v.walk(fn) // cannot fail: the view was validated
362}
363
364func (v snapshotView) walk(fn func(key, value []byte, version uint64)) error {
365 body := v.body
366 var prev []byte
367 for i := uint64(0); i < v.count; i++ {
368 var key, value []byte
369 var ok bool
370 if key, body, ok = cutBytes(body, maxKeyLen); !ok || len(key) == 0 {
371 return fmt.Errorf("%w: bad snapshot key at entry %d", ErrCorrupt, i)
372 }
373 if i > 0 && bytes.Compare(prev, key) >= 0 {
374 return fmt.Errorf("%w: snapshot keys not ascending at entry %d", ErrCorrupt, i)
375 }
376 version, n := binary.Uvarint(body)
377 if n <= 0 || version > v.seq {
378 return fmt.Errorf("%w: bad snapshot version at entry %d", ErrCorrupt, i)
379 }
380 if value, body, ok = cutBytes(body[n:], maxRecord); !ok {
381 return fmt.Errorf("%w: bad snapshot value at entry %d", ErrCorrupt, i)
382 }
383 if fn != nil {
384 fn(key, value, version)
385 }
386 prev = key
387 }
388 if len(body) != 0 {
389 return fmt.Errorf("%w: %d trailing snapshot bytes", ErrCorrupt, len(body))
390 }
391 return nil
392}