A very experimental PLC implementation which uses BFT consensus for decentralization
Rename store.Tree to store.Consensus
+42
-42
+2
-2
abciapp/app.go
+2
-2
abciapp/app.go
···
71
71
aocsByPLC: make(map[string]*authoritativeOperationsCache),
72
72
}
73
73
74
-
d.txFactory, err = transaction.NewFactory(tree, indexDB, store.Tree.NextOperationSequence, store.NewDIDBloomFilterStore(didBloomFilterPath))
74
+
d.txFactory, err = transaction.NewFactory(tree, indexDB, store.Consensus.NextOperationSequence, store.NewDIDBloomFilterStore(didBloomFilterPath))
75
75
if err != nil {
76
76
return nil, nil, nil, func() {}, stacktrace.Propagate(err, "")
77
77
}
···
88
88
89
89
*d.tree = *mkTree()
90
90
91
-
d.txFactory, err = transaction.NewFactory(tree, indexDB, store.Tree.NextOperationSequence, store.NewDIDBloomFilterStore(didBloomFilterPath))
91
+
d.txFactory, err = transaction.NewFactory(tree, indexDB, store.Consensus.NextOperationSequence, store.NewDIDBloomFilterStore(didBloomFilterPath))
92
92
if err != nil {
93
93
return stacktrace.Propagate(err, "")
94
94
}
+2
-2
abciapp/import.go
+2
-2
abciapp/import.go
···
251
251
// use WorkingTreeVersion so we take into account any import operation that may have been processed in this block
252
252
readTx := d.txFactory.ReadWorking(time.Now())
253
253
254
-
plcURL, err := store.Tree.AuthoritativePLC(readTx)
254
+
plcURL, err := store.Consensus.AuthoritativePLC(readTx)
255
255
if err != nil {
256
256
return nil, stacktrace.Propagate(err, "")
257
257
}
···
261
261
return nil, nil
262
262
}
263
263
264
-
cursor, err := store.Tree.AuthoritativeImportProgress(readTx)
264
+
cursor, err := store.Consensus.AuthoritativeImportProgress(readTx)
265
265
if err != nil {
266
266
return nil, stacktrace.Propagate(err, "")
267
267
}
+5
-5
abciapp/tx_import.go
+5
-5
abciapp/tx_import.go
···
54
54
}
55
55
56
56
if writeTx, ok := deps.writeTx.Get(); ok {
57
-
err = store.Tree.SetAuthoritativePLC(writeTx, tx.Arguments.PLCURL)
57
+
err = store.Consensus.SetAuthoritativePLC(writeTx, tx.Arguments.PLCURL)
58
58
if err != nil {
59
59
return nil, stacktrace.Propagate(err, "")
60
60
}
61
61
62
62
if tx.Arguments.RestartImport {
63
-
err = store.Tree.SetAuthoritativeImportProgress(writeTx, 0)
63
+
err = store.Consensus.SetAuthoritativeImportProgress(writeTx, 0)
64
64
if err != nil {
65
65
return nil, stacktrace.Propagate(err, "")
66
66
}
···
99
99
}, nil
100
100
}
101
101
102
-
expectedPlcUrl, err := store.Tree.AuthoritativePLC(deps.readTx)
102
+
expectedPlcUrl, err := store.Consensus.AuthoritativePLC(deps.readTx)
103
103
if err != nil {
104
104
return nil, stacktrace.Propagate(err, "")
105
105
}
···
113
113
114
114
aoc := getOrCreateAuthoritativeOperationsCache(deps.runnerContext, deps.aocsByPLC, expectedPlcUrl)
115
115
116
-
expectedCursor, err := store.Tree.AuthoritativeImportProgress(deps.readTx)
116
+
expectedCursor, err := store.Consensus.AuthoritativeImportProgress(deps.readTx)
117
117
if err != nil {
118
118
return nil, stacktrace.Propagate(err, "")
119
119
}
···
165
165
}
166
166
}
167
167
168
-
err = store.Tree.SetAuthoritativeImportProgress(writeTx, newCursor)
168
+
err = store.Consensus.SetAuthoritativeImportProgress(writeTx, newCursor)
169
169
if err != nil {
170
170
return nil, stacktrace.Propagate(err, "")
171
171
}
+11
-11
plc/impl.go
+11
-11
plc/impl.go
···
47
47
return stacktrace.Propagate(err, "operation failed validation")
48
48
}
49
49
50
-
err = store.Tree.StoreOperation(ctx, tx, effects.NewLogEntry, effects.NullifiedEntriesStartingSeq)
50
+
err = store.Consensus.StoreOperation(ctx, tx, effects.NewLogEntry, effects.NullifiedEntriesStartingSeq)
51
51
if err != nil {
52
52
return stacktrace.Propagate(err, "failed to commit operation")
53
53
}
···
63
63
nullifiedEntriesStartingSeq := mo.None[uint64]()
64
64
65
65
var iteratorErr error
66
-
for entry := range store.Tree.AuditLogReverseIterator(ctx, tx.Downgrade(), newEntry.DID, &iteratorErr) {
66
+
for entry := range store.Consensus.AuditLogReverseIterator(ctx, tx.Downgrade(), newEntry.DID, &iteratorErr) {
67
67
entryCID := entry.CID.String()
68
68
if !hasExistingOps {
69
69
hasExistingOps = true
···
84
84
}
85
85
86
86
return stacktrace.Propagate(
87
-
store.Tree.SetOperationCreatedAt(tx, entry.Seq, newCreatedAtDT.Time()),
87
+
store.Consensus.SetOperationCreatedAt(tx, entry.Seq, newCreatedAtDT.Time()),
88
88
"")
89
89
}
90
90
···
110
110
// there's nothing to do but store the operation, no nullification involved
111
111
newEntry.Nullified = false
112
112
113
-
err := store.Tree.StoreOperation(ctx, tx, newEntry, nullifiedEntriesStartingSeq)
113
+
err := store.Consensus.StoreOperation(ctx, tx, newEntry, nullifiedEntriesStartingSeq)
114
114
return stacktrace.Propagate(err, "failed to commit operation")
115
115
}
116
116
···
123
123
}
124
124
125
125
newEntry.Nullified = false
126
-
err := store.Tree.StoreOperation(ctx, tx, newEntry, nullifiedEntriesStartingSeq)
126
+
err := store.Consensus.StoreOperation(ctx, tx, newEntry, nullifiedEntriesStartingSeq)
127
127
return stacktrace.Propagate(err, "failed to commit operation")
128
128
}
129
129
130
130
func (plc *plcImpl) Resolve(ctx context.Context, tx transaction.Read, did string) (didplc.Doc, error) {
131
131
var iteratorErr error
132
-
for entry := range store.Tree.AuditLogReverseIterator(ctx, tx, did, &iteratorErr) {
132
+
for entry := range store.Consensus.AuditLogReverseIterator(ctx, tx, did, &iteratorErr) {
133
133
if entry.Operation.Tombstone != nil {
134
134
return didplc.Doc{}, stacktrace.Propagate(ErrDIDGone, "")
135
135
}
···
147
147
// if missing -> returns ErrDIDNotFound
148
148
// if tombstone -> returns log as normal
149
149
150
-
l, _, err := store.Tree.AuditLog(ctx, tx, did, false)
150
+
l, _, err := store.Consensus.AuditLog(ctx, tx, did, false)
151
151
if err != nil {
152
152
return nil, stacktrace.Propagate(err, "")
153
153
}
···
169
169
// GetPlcAuditLog - /:did/log/audit - full audit log, with nullified
170
170
// if missing -> returns ErrDIDNotFound
171
171
// if tombstone -> returns log as normal
172
-
l, _, err := store.Tree.AuditLog(ctx, tx, did, false)
172
+
l, _, err := store.Consensus.AuditLog(ctx, tx, did, false)
173
173
if err != nil {
174
174
return nil, stacktrace.Propagate(err, "")
175
175
}
···
189
189
// if tombstone -> returns tombstone op
190
190
191
191
var iteratorErr error
192
-
for entry := range store.Tree.AuditLogReverseIterator(ctx, tx, did, &iteratorErr) {
192
+
for entry := range store.Consensus.AuditLogReverseIterator(ctx, tx, did, &iteratorErr) {
193
193
return entry.Operation, nil
194
194
}
195
195
if iteratorErr != nil {
···
205
205
// if tombstone -> returns ErrDIDGone
206
206
207
207
var iteratorErr error
208
-
for entry := range store.Tree.AuditLogReverseIterator(ctx, tx, did, &iteratorErr) {
208
+
for entry := range store.Consensus.AuditLogReverseIterator(ctx, tx, did, &iteratorErr) {
209
209
opEnum := entry.Operation
210
210
if opEnum.Tombstone != nil {
211
211
return didplc.RegularOp{}, stacktrace.Propagate(ErrDIDGone, "")
···
223
223
}
224
224
225
225
func (plc *plcImpl) Export(ctx context.Context, tx transaction.Read, after uint64, count int) ([]types.SequencedLogEntry, error) {
226
-
entries, err := store.Tree.ExportOperations(ctx, tx, after, count)
226
+
entries, err := store.Consensus.ExportOperations(ctx, tx, after, count)
227
227
return entries, stacktrace.Propagate(err, "")
228
228
}
+2
-2
plc/operation_validator.go
+2
-2
plc/operation_validator.go
···
78
78
nullifiedEntriesStartingSeq := mo.None[uint64]()
79
79
80
80
var iteratorErr error
81
-
for entry := range store.Tree.AuditLogReverseIterator(ctx, readTx, expectedDid, &iteratorErr) {
81
+
for entry := range store.Consensus.AuditLogReverseIterator(ctx, readTx, expectedDid, &iteratorErr) {
82
82
relevantExistingEntries = append(relevantExistingEntries, entry)
83
83
if !hasExistingOps {
84
84
hasExistingOps = true
···
218
218
219
219
var withinHour, withinDay, withinWeek int
220
220
var err error
221
-
for entry := range store.Tree.AuditLogReverseIterator(ctx, tx, did, &err) {
221
+
for entry := range store.Consensus.AuditLogReverseIterator(ctx, tx, did, &err) {
222
222
if entry.Nullified {
223
223
// The typescript implementation operates over a `ops` array which doesn't include nullified ops
224
224
// (With recovery ops also skipping rate limits, doesn't this leave the PLC vulnerable to the spam of constant recovery operations? TODO investigate)
+1
-1
store/did_bloom.go
+1
-1
store/did_bloom.go
···
72
72
73
73
// update bloom filter with DIDs it may have missed since the time it was serialized
74
74
var iterErr error
75
-
for entry := range Tree.OperationsIterator(tx, max(1, seq)-1, &iterErr) {
75
+
for entry := range Consensus.OperationsIterator(tx, max(1, seq)-1, &iterErr) {
76
76
didBytes, err := DIDToBytes(entry.DID)
77
77
if err != nil {
78
78
return nil, 0, stacktrace.Propagate(err, "")
+18
-18
store/tree.go
store/consensus.go
+18
-18
store/tree.go
store/consensus.go
···
24
24
"tangled.org/gbl08ma.com/didplcbft/types"
25
25
)
26
26
27
-
// TODO rename to something more appropriate, now that this touches both the tree and the index
28
-
var Tree *TreeStore = &TreeStore{}
27
+
var Consensus ConsensusStore = &consensusStore{}
29
28
30
-
type PLCTreeStore interface {
29
+
// ConsensusStore manages all information that is directly or indirectly protected by consensus
30
+
type ConsensusStore interface {
31
31
AuditLog(ctx context.Context, tx transaction.Read, did string, withProof bool) ([]types.SequencedLogEntry, *ics23.CommitmentProof, error)
32
32
AuditLogReverseIterator(ctx context.Context, tx transaction.Read, did string, err *error) iter.Seq[types.SequencedLogEntry]
33
33
···
46
46
SetAuthoritativeImportProgress(tx transaction.Write, nextCursor uint64) error
47
47
}
48
48
49
-
var _ PLCTreeStore = (*TreeStore)(nil)
49
+
var _ ConsensusStore = (*consensusStore)(nil)
50
50
51
-
// TreeStore exists just to groups methods nicely
52
-
type TreeStore struct{}
51
+
// consensusStore exists just to groups methods nicely
52
+
type consensusStore struct{}
53
53
54
-
func (t *TreeStore) AuditLog(ctx context.Context, tx transaction.Read, did string, withProof bool) ([]types.SequencedLogEntry, *ics23.CommitmentProof, error) {
54
+
func (t *consensusStore) AuditLog(ctx context.Context, tx transaction.Read, did string, withProof bool) ([]types.SequencedLogEntry, *ics23.CommitmentProof, error) {
55
55
didBytes, err := DIDToBytes(did)
56
56
if err != nil {
57
57
return nil, nil, stacktrace.Propagate(err, "")
···
121
121
return logEntries, combinedProof, nil
122
122
}
123
123
124
-
func (t *TreeStore) AuditLogReverseIterator(ctx context.Context, tx transaction.Read, did string, retErr *error) iter.Seq[types.SequencedLogEntry] {
124
+
func (t *consensusStore) AuditLogReverseIterator(ctx context.Context, tx transaction.Read, did string, retErr *error) iter.Seq[types.SequencedLogEntry] {
125
125
return func(yield func(types.SequencedLogEntry) bool) {
126
126
didBytes, err := DIDToBytes(did)
127
127
if err != nil {
···
192
192
}
193
193
}
194
194
195
-
func (t *TreeStore) ExportOperations(ctx context.Context, tx transaction.Read, after uint64, count int) ([]types.SequencedLogEntry, error) {
195
+
func (t *consensusStore) ExportOperations(ctx context.Context, tx transaction.Read, after uint64, count int) ([]types.SequencedLogEntry, error) {
196
196
entries := make([]types.SequencedLogEntry, 0, count)
197
197
var iterErr error
198
198
for logEntry := range t.OperationsIterator(tx, after, &iterErr) {
···
209
209
return entries, nil
210
210
}
211
211
212
-
func (t *TreeStore) OperationsIterator(tx transaction.Read, after uint64, retErr *error) iter.Seq[types.SequencedLogEntry] {
212
+
func (t *consensusStore) OperationsIterator(tx transaction.Read, after uint64, retErr *error) iter.Seq[types.SequencedLogEntry] {
213
213
return func(yield func(types.SequencedLogEntry) bool) {
214
214
// as the name suggests, after is an exclusive lower bound, but our iterators use inclusive lower bounds
215
215
start := after + 1
···
249
249
//
250
250
// Even though this function is not meant to overwrite operations (it will error) we ask the caller to provide the sequence
251
251
// The caller is responsible for managing the sequence and invalidating it when needed (e.g. after a rollback) using
252
-
// [[TreeStore.NextOperationSequence]].
252
+
// [[consensusStore.NextOperationSequence]].
253
253
// Pushing the responsibility to the caller is preferable in terms of performance, even if it leads to less safe code,
254
254
// because getting the sequence from the tree within this function every time has a significant performance hit
255
-
func (t *TreeStore) StoreOperation(ctx context.Context, tx transaction.Write, entry didplc.LogEntry, nullifyWithSequenceEqualOrGreaterThan mo.Option[uint64]) error {
255
+
func (t *consensusStore) StoreOperation(ctx context.Context, tx transaction.Write, entry didplc.LogEntry, nullifyWithSequenceEqualOrGreaterThan mo.Option[uint64]) error {
256
256
didBytes, err := DIDToBytes(entry.DID)
257
257
if err != nil {
258
258
return stacktrace.Propagate(err, "")
···
354
354
return nil
355
355
}
356
356
357
-
func (t *TreeStore) SetOperationCreatedAt(tx transaction.Write, seqID uint64, createdAt time.Time) error {
357
+
func (t *consensusStore) SetOperationCreatedAt(tx transaction.Write, seqID uint64, createdAt time.Time) error {
358
358
opKey := marshalOperationKey(seqID)
359
359
360
360
opValue, err := tx.Tree().Get(opKey)
···
382
382
var minOperationKey = marshalOperationKey(0)
383
383
var maxOperationKey = marshalOperationKey(math.MaxInt64)
384
384
385
-
func (t *TreeStore) NextOperationSequence(tx transaction.Read) (uint64, error) {
385
+
func (t *consensusStore) NextOperationSequence(tx transaction.Read) (uint64, error) {
386
386
seq := uint64(0)
387
387
388
388
itr, err := tx.Tree().Iterator(minOperationKey, maxOperationKey, false)
···
585
585
Complete())
586
586
}
587
587
588
-
func (t *TreeStore) AuthoritativePLC(tx transaction.Read) (string, error) {
588
+
func (t *consensusStore) AuthoritativePLC(tx transaction.Read) (string, error) {
589
589
url, err := tx.Tree().Get([]byte("aPLCURL"))
590
590
if err != nil {
591
591
return "", stacktrace.Propagate(err, "")
···
596
596
return string(url), nil
597
597
}
598
598
599
-
func (t *TreeStore) SetAuthoritativePLC(tx transaction.Write, url string) error {
599
+
func (t *consensusStore) SetAuthoritativePLC(tx transaction.Write, url string) error {
600
600
_, err := tx.Tree().Set([]byte("aPLCURL"), []byte(url))
601
601
return stacktrace.Propagate(err, "")
602
602
}
603
603
604
-
func (t *TreeStore) AuthoritativeImportProgress(tx transaction.Read) (uint64, error) {
604
+
func (t *consensusStore) AuthoritativeImportProgress(tx transaction.Read) (uint64, error) {
605
605
progBytes, err := tx.Tree().Get([]byte("aImportProgress"))
606
606
if err != nil {
607
607
return 0, stacktrace.Propagate(err, "")
···
612
612
return binary.BigEndian.Uint64(progBytes), nil
613
613
}
614
614
615
-
func (t *TreeStore) SetAuthoritativeImportProgress(tx transaction.Write, nextCursor uint64) error {
615
+
func (t *consensusStore) SetAuthoritativeImportProgress(tx transaction.Write, nextCursor uint64) error {
616
616
value := make([]byte, 8)
617
617
binary.BigEndian.PutUint64(value, nextCursor)
618
618
+1
-1
testutil/testutil.go
+1
-1
testutil/testutil.go
···
23
23
_, indexDB, err := badgertodbm.NewBadgerInMemoryDB()
24
24
require.NoError(t, err)
25
25
26
-
factory, err := transaction.NewFactory(tree, indexDB, store.Tree.NextOperationSequence, store.NewInMemoryDIDBloomFilterStore())
26
+
factory, err := transaction.NewFactory(tree, indexDB, store.Consensus.NextOperationSequence, store.NewInMemoryDIDBloomFilterStore())
27
27
require.NoError(t, err)
28
28
29
29
return factory, tree, indexDB