A very experimental PLC implementation which uses BFT consensus for decentralization
Compress tree values using zstd with custom dictionary
+57
-5
+3
-2
badger.go
+3
-2
badger.go
···
38
38
opts.NumCompactors = 8
39
39
40
40
// use low ValueThreshold to force operation values into the value logs
41
-
// sadly this means they won't be compressed at all (TODO implement compression ourselves, maybe based on a fixed zstd dictionary?)
41
+
// sadly this means they won't be compressed at all, so we're compressing the operations ourselves
42
+
// (with the benefit of being able to use a custom dictionary for much better compression ratios)
42
43
// because these large values don't seem to change (unlike the non-leaf nodes within the iavl tree, which are changing all the time),
43
44
// this makes compaction much faster and decreases SSD thrashing
44
-
opts.ValueThreshold = 256
45
+
opts.ValueThreshold = 192
45
46
return NewBadgerDBWithOptions(opts)
46
47
}
47
48
+54
-3
store/tree.go
+54
-3
store/tree.go
···
2
2
3
3
import (
4
4
"context"
5
+
_ "embed"
5
6
"encoding/base32"
6
7
"encoding/binary"
7
8
"iter"
···
14
15
ics23 "github.com/cosmos/ics23/go"
15
16
"github.com/did-method-plc/go-didplc"
16
17
cbornode "github.com/ipfs/go-ipld-cbor"
18
+
"github.com/klauspost/compress/zstd"
17
19
"github.com/palantir/stacktrace"
18
20
"github.com/polydawn/refmt/obj/atlas"
19
21
"github.com/samber/lo"
···
39
41
40
42
AuthoritativeImportProgress(tx transaction.Read) (uint64, error)
41
43
SetAuthoritativeImportProgress(tx transaction.Write, nextCursor uint64) error
44
+
45
+
ProduceOperationExamples(tx transaction.Read, interval, count int) iter.Seq[[]byte]
42
46
}
43
47
44
48
var _ PLCTreeStore = (*TreeStore)(nil)
···
46
50
// TreeStore exists just to groups methods nicely
47
51
type TreeStore struct{}
48
52
53
+
//go:embed zstddict/plcops
54
+
var plcOpsZstdDict []byte
55
+
56
+
// Using SpeedDefault appears to cause the processing time for ExecuteOperation to double on average
57
+
// Using SpeedBetterCompression appears to cause the processing time to double again
58
+
// By using SpeedFastest we seem to give up on like 5% size reduction, it's not worth using the slower speeds
59
+
var zstdOpEncoder, _ = zstd.NewWriter(nil, zstd.WithEncoderDict(plcOpsZstdDict), zstd.WithEncoderLevel(zstd.SpeedFastest))
60
+
var zstdOpDecoder, _ = zstd.NewReader(nil, zstd.WithDecoderDicts(plcOpsZstdDict))
61
+
62
+
func (t *TreeStore) ProduceOperationExamples(tx transaction.Read, interval, count int) iter.Seq[[]byte] {
63
+
return func(yield func([]byte) bool) {
64
+
for i := 0; i < count*interval; i += interval {
65
+
key := marshalOperationKey(uint64(i))
66
+
67
+
value, err := tx.Tree().Get(key)
68
+
if err != nil {
69
+
continue
70
+
}
71
+
72
+
if !yield(slices.Clone(value)) {
73
+
return
74
+
}
75
+
}
76
+
}
77
+
78
+
}
79
+
49
80
func (t *TreeStore) AuditLog(ctx context.Context, tx transaction.Read, did string, withProof bool) ([]types.SequencedLogEntry, *ics23.CommitmentProof, error) {
50
81
didBytes, err := DIDToBytes(did)
51
82
if err != nil {
···
259
290
if err != nil {
260
291
return stacktrace.Propagate(err, "")
261
292
}
262
-
operationValue = slices.Clone(operationValue)
263
-
operationValue[0] = 1
293
+
operationValue, err = markCompressedOperationValueNullified(operationValue)
294
+
if err != nil {
295
+
return stacktrace.Propagate(err, "")
296
+
}
264
297
265
298
updated, err := tx.Tree().Set(opKey, operationValue)
266
299
if err != nil {
···
444
477
binary.BigEndian.PutUint64(o[16:24], ts)
445
478
copy(o[24:], opAsBytes)
446
479
447
-
return o
480
+
return zstdOpEncoder.EncodeAll(o, make([]byte, 0, len(o)))
448
481
}
449
482
450
483
func unmarshalOperationValue(value []byte) (bool, string, time.Time, didplc.OpEnum, error) {
484
+
// passing a nil output buffer to DecodeAll means it'll optimistically start by allocating len(value)*2
485
+
// but we observe compression ratios better than 50% sometimes, so allocate len(value)*3 instead
486
+
value, err := zstdOpDecoder.DecodeAll(value, make([]byte, 0, len(value)*3))
487
+
if err != nil {
488
+
return false, "", time.Time{}, didplc.OpEnum{}, stacktrace.Propagate(err, "")
489
+
}
490
+
451
491
nullified := value[0] != 0
452
492
453
493
did, err := bytesToDID(value[1:16])
···
464
504
return false, "", time.Time{}, didplc.OpEnum{}, stacktrace.Propagate(err, "")
465
505
}
466
506
return nullified, did, createdAt, opEnum, nil
507
+
}
508
+
509
+
func markCompressedOperationValueNullified(value []byte) ([]byte, error) {
510
+
value, err := zstdOpDecoder.DecodeAll(value, make([]byte, 0, len(value)*3))
511
+
if err != nil {
512
+
return nil, stacktrace.Propagate(err, "")
513
+
}
514
+
515
+
value[0] = 1
516
+
517
+
return zstdOpEncoder.EncodeAll(value, make([]byte, 0, len(value))), nil
467
518
}
468
519
469
520
func unmarshalLogEntry(operationKey, operationValue []byte) (types.SequencedLogEntry, error) {
store/zstddict/plcops
store/zstddict/plcops
This is a binary file and will not be displayed.