+23 -7

httpapi/server.go

··· 9 9 "net" 10 10 "net/http" 11 11 "slices" 12 12 + "strconv" 12 13 "strings" 13 14 "sync" 14 15 "sync/atomic" 15 16 "time" 16 17 17 18 "github.com/bluesky-social/indigo/atproto/atcrypto" 18 18 - "github.com/bluesky-social/indigo/atproto/syntax" 19 19 "github.com/cometbft/cometbft/node" 20 20 "github.com/did-method-plc/go-didplc" 21 21 "github.com/google/uuid" 22 22 cbornode "github.com/ipfs/go-ipld-cbor" 23 23 "github.com/palantir/stacktrace" 24 24 "github.com/rs/cors" 25 25 + "github.com/samber/lo" 25 26 26 27 "tangled.org/gbl08ma/didplcbft/abciapp" 27 28 "tangled.org/gbl08ma/didplcbft/plc" ··· 307 308 } 308 309 309 310 afterStr := query.Get("after") 310 310 - after, err := syntax.ParseDatetime(afterStr) 311 311 - if err != nil && afterStr != "" { 312 312 - sendErrorResponse(w, http.StatusBadRequest, "Invalid after parameter") 313 313 - return 311 311 + after := uint64(0) 312 312 + var err error 313 313 + if afterStr != "" { 314 314 + after, err = strconv.ParseUint(afterStr, 10, 64) 315 315 + if err != nil { 316 316 + sendErrorResponse(w, http.StatusBadRequest, "Only sequence-based pagination is supported") 317 317 + return 318 318 + } 314 319 } 315 320 316 316 - entries, err := s.plc.Export(r.Context(), plc.CommittedTreeVersion, after.Time(), count) 321 321 + entries, err := s.plc.Export(r.Context(), plc.CommittedTreeVersion, after, count) 317 322 if handlePLCError(w, err, "") { 318 323 return 319 324 } 320 325 321 326 w.Header().Set("Content-Type", "application/jsonlines") 327 327 + 328 328 + type jsonEntry struct { 329 329 + Seq uint64 `json:"seq"` 330 330 + Type string `json:"type"` 331 331 + *didplc.LogEntry 332 332 + } 322 333 for _, entry := range entries { 323 323 - json.NewEncoder(w).Encode(&entry) 334 334 + json.NewEncoder(w).Encode(jsonEntry{ 335 335 + Seq: entry.Seq, 336 336 + Type: "sequenced_op", 337 337 + LogEntry: lo.ToPtr(entry.ToDIDPLCLogEntry()), 338 338 + }) 324 339 } 325 340 } 326 341 ··· 342 357 343 358 // sendErrorResponse sends an error response with the specified status code and message. 344 359 func sendErrorResponse(w http.ResponseWriter, statusCode int, message string) { 360 360 + w.Header().Set("Content-Type", "application/json") 345 361 w.WriteHeader(statusCode) 346 362 json.NewEncoder(w).Encode(map[string]string{"message": message}) 347 363 }

+4 -3

httpapi/server_test.go

··· 13 13 "github.com/did-method-plc/go-didplc" 14 14 "github.com/stretchr/testify/require" 15 15 "tangled.org/gbl08ma/didplcbft/plc" 16 16 + "tangled.org/gbl08ma/didplcbft/types" 16 17 ) 17 18 18 19 // MockReadPLC is a mock implementation of the ReadPLC interface for testing. ··· 92 93 return didplc.RegularOp{}, nil 93 94 } 94 95 95 95 - func (m *MockReadPLC) Export(ctx context.Context, atHeight plc.TreeVersion, after time.Time, count int) ([]didplc.LogEntry, error) { 96 96 + func (m *MockReadPLC) Export(ctx context.Context, atHeight plc.TreeVersion, after uint64, count int) ([]types.SequencedLogEntry, error) { 96 97 if m.shouldReturnError { 97 97 - return []didplc.LogEntry{}, fmt.Errorf("internal error") 98 98 + return []types.SequencedLogEntry{}, fmt.Errorf("internal error") 98 99 } 99 99 - return []didplc.LogEntry{}, nil 100 100 + return []types.SequencedLogEntry{}, nil 100 101 } 101 102 102 103 func TestServer(t *testing.T) {

+14 -23

importer/importer_test.go

··· 25 25 ) 26 26 27 27 func TestImportV2(t *testing.T) { 28 28 - c, err := rpchttp.New("http://localhost:26100", "/websocket") 28 28 + c, err := rpchttp.New("http://localhost:26657", "/websocket") 29 29 require.NoError(t, err) 30 30 31 31 ctx := t.Context() ··· 41 41 var wg sync.WaitGroup 42 42 noMoreNewEntries := atomic.Bool{} 43 43 wg.Go(func() { 44 44 - for entry := range iterateOverExport(ctx, "2023-10-10T00:00:00.000Z") { 44 44 + for entry := range iterateOverExport(ctx, 0) { 45 45 if totalAwaiting.Size() > 5000 { 46 46 for totalAwaiting.Size() > 1000 { 47 47 time.Sleep(1 * time.Second) ··· 184 184 wg.Wait() 185 185 } 186 186 187 187 - func iterateOverExport(ctx context.Context, startAt string) iter.Seq[didplc.LogEntry] { 187 187 + func iterateOverExport(ctx context.Context, startAt uint64) iter.Seq[didplc.LogEntry] { 188 188 return func(yield func(didplc.LogEntry) bool) { 189 189 const batchSize = 1000 190 190 baseURL := didplc.DefaultDirectoryURL + "/export" 191 191 client := &http.Client{Timeout: 30 * time.Second} 192 192 - 193 193 - // The /export seems to sometimes return outright duplicated entries :weary: 194 194 - seenCIDs := map[string]struct{}{} 195 192 196 193 after := startAt 197 194 for { ··· 204 201 205 202 q := req.URL.Query() 206 203 q.Add("count", fmt.Sprint(batchSize)) 207 207 - if after != "" { 208 208 - q.Add("after", after) 209 209 - } 204 204 + q.Add("after", fmt.Sprint(after)) 210 205 req.URL.RawQuery = q.Encode() 211 206 212 207 resp, err := client.Do(req) ··· 219 214 return // Non-200 status code 220 215 } 221 216 222 222 - entries := make([]didplc.LogEntry, 0, batchSize) 217 217 + type logEntryWithSeq struct { 218 218 + didplc.LogEntry 219 219 + Seq uint64 `json:"seq"` 220 220 + } 221 221 + 222 222 + entries := make([]logEntryWithSeq, 0, batchSize) 223 223 224 224 // Read response body 225 225 s := bufio.NewScanner(resp.Body) 226 226 receivedEntries := 0 227 227 for s.Scan() { 228 228 - var entry didplc.LogEntry 228 228 + var entry logEntryWithSeq 229 229 if err := json.Unmarshal(s.Bytes(), &entry); err != nil { 230 230 return // Failed to decode JSON 231 231 } 232 232 - if _, present := seenCIDs[entry.CID]; !present { 233 233 - entries = append(entries, entry) 234 234 - seenCIDs[entry.CID] = struct{}{} 235 235 - } 232 232 + entries = append(entries, entry) 236 233 receivedEntries++ 237 234 } 238 235 if s.Err() != nil { ··· 244 241 } 245 242 246 243 // Process each entry 247 247 - var lastCreatedAt string 248 244 for _, entry := range entries { 249 249 - lastCreatedAt = entry.CreatedAt 250 250 - if !yield(entry) { 245 245 + after = entry.Seq 246 246 + if !yield(entry.LogEntry) { 251 247 return 252 248 } 253 249 } ··· 255 251 if receivedEntries < batchSize { 256 252 return 257 253 } 258 258 - 259 259 - after = lastCreatedAt 260 260 - 261 261 - // Small delay to be respectful to the API 262 262 - time.Sleep(100 * time.Millisecond) 263 254 } 264 255 } 265 256 }

+66 -28

plc/impl.go

··· 13 13 "github.com/samber/lo" 14 14 "github.com/samber/mo" 15 15 "tangled.org/gbl08ma/didplcbft/store" 16 16 + "tangled.org/gbl08ma/didplcbft/types" 16 17 ) 17 18 18 19 type TreeProvider interface { ··· 43 44 plc.mu.Lock() 44 45 defer plc.mu.Unlock() 45 46 46 46 - timestamp := syntax.Datetime(at.Format(store.ActualAtprotoDatetimeLayout)) 47 47 + timestamp := syntax.Datetime(at.Format(types.ActualAtprotoDatetimeLayout)) 47 48 48 49 // TODO set true to false only while importing old ops 49 50 _, err := plc.validator.Validate(atHeight, timestamp, did, opBytes, true) ··· 58 59 plc.mu.Lock() 59 60 defer plc.mu.Unlock() 60 61 61 61 - timestamp := syntax.Datetime(t.Format(store.ActualAtprotoDatetimeLayout)) 62 62 + timestamp := syntax.Datetime(t.Format(types.ActualAtprotoDatetimeLayout)) 62 63 63 64 // TODO set true to false only while importing old ops 64 65 effects, err := plc.validator.Validate(WorkingTreeVersion, timestamp, did, opBytes, true) ··· 97 98 newCID := newEntry.CID 98 99 newPrev := newEntry.Operation.AsOperation().PrevCIDStr() 99 100 100 100 - // TODO avoid redundant CreatedAt formating and parsing by using a specialized LogEntry type internally (i.e. between us and the store) 101 101 newCreatedAtDT, err := syntax.ParseDatetime(newEntry.CreatedAt) 102 102 if err != nil { 103 103 return stacktrace.Propagate(err, "") ··· 106 106 107 107 mustFullyReplaceHistory := false 108 108 for _, entry := range l { 109 109 - existingCreatedAt, err := syntax.ParseDatetime(entry.CreatedAt) 110 110 - if err != nil { 111 111 - return stacktrace.Propagate(err, "") 112 112 - } 113 113 - if existingCreatedAt.Time().After(newCreatedAt) { 109 109 + if entry.CreatedAt.After(newCreatedAt) { 114 110 // We're trying to import an operation whose timestamp precedes one of the timestamps for operations we already know about 115 111 // We'll need to discard all known history and import it anew using the authoritative source data (same as when dealing with sequence forks) 116 112 mustFullyReplaceHistory = true 117 113 break 118 114 } 119 115 120 120 - if entry.CID == newCID { 116 116 + if entry.CID.String() == newCID && entry.Nullified == newEntry.Nullified { 121 117 // If an operation with the same CID already exists -> easy-ish 122 118 123 119 // this operation is already present, there is nothing to do ··· 127 123 } 128 124 } 129 125 130 130 - if len(l) == 0 || (!mustFullyReplaceHistory && l[len(l)-1].CID == newPrev) { 126 126 + if len(l) == 0 || (!mustFullyReplaceHistory && l[len(l)-1].CID.String() == newPrev) { 131 127 // If DID doesn't exist at all -> easy 132 128 // If prev matches CID of latest operation, and resulting timestamp sequence monotonically increases -> easy 133 129 err = store.Tree.StoreOperation(tree, newEntry, mo.None[int]()) ··· 166 162 return didplc.Doc{}, stacktrace.Propagate(ErrDIDNotFound, "") 167 163 } 168 164 169 169 - opEnum := l[len(l)-1].Operation 170 170 - if opEnum.Tombstone != nil { 171 171 - return didplc.Doc{}, stacktrace.Propagate(ErrDIDGone, "") 165 165 + // find most recent operation that isn't nullified (during authoritative import, the latest operation might be nullified) 166 166 + for i := len(l) - 1; i >= 0; i-- { 167 167 + opEnum := l[i].Operation 168 168 + if !l[i].Nullified { 169 169 + if opEnum.Tombstone != nil { 170 170 + return didplc.Doc{}, stacktrace.Propagate(ErrDIDGone, "") 171 171 + } 172 172 + return opEnum.AsOperation().Doc(did) 173 173 + } 172 174 } 173 173 - return opEnum.AsOperation().Doc(did) 175 175 + // in the worst case all operations are somehow nullified and the loop ends with opEnum holding a nullified operation 176 176 + // that _shouldn't_ be possible (right?) but if it does happen, let's just behave as if the DID was tombstoned 177 177 + return didplc.Doc{}, stacktrace.Propagate(ErrDIDGone, "") 174 178 } 175 179 176 180 func (plc *plcImpl) OperationLog(ctx context.Context, atHeight TreeVersion, did string) ([]didplc.OpEnum, error) { ··· 195 199 return nil, stacktrace.Propagate(ErrDIDNotFound, "") 196 200 } 197 201 198 198 - return lo.Map(l, func(logEntry didplc.LogEntry, _ int) didplc.OpEnum { 202 202 + l = lo.Filter(l, func(logEntry types.SequencedLogEntry, _ int) bool { 203 203 + return !logEntry.Nullified 204 204 + }) 205 205 + 206 206 + return lo.Map(l, func(logEntry types.SequencedLogEntry, _ int) didplc.OpEnum { 199 207 return logEntry.Operation 200 208 }), nil 201 209 } ··· 221 229 return nil, stacktrace.Propagate(ErrDIDNotFound, "") 222 230 } 223 231 224 224 - return l, nil 232 232 + // if the latest operations are nullified (happens while authoritative import is in progress), just pretend we don't have them yet, 233 233 + // since a properly functioning PLC implementation could never have the latest operation for a DID be nullified 234 234 + dropAfterIdx := len(l) - 1 235 235 + for ; dropAfterIdx >= 0; dropAfterIdx-- { 236 236 + if !l[dropAfterIdx].Nullified { 237 237 + break 238 238 + } 239 239 + } 240 240 + l = l[0 : dropAfterIdx+1] 241 241 + 242 242 + return lo.Map(l, func(logEntry types.SequencedLogEntry, _ int) didplc.LogEntry { 243 243 + return logEntry.ToDIDPLCLogEntry() 244 244 + }), nil 225 245 } 226 246 227 247 func (plc *plcImpl) LastOperation(ctx context.Context, atHeight TreeVersion, did string) (didplc.OpEnum, error) { 228 228 - // GetLastOp - /:did/log/last - latest op from audit log which isn't nullified (isn't the latest op guaranteed to not be nullified?) 248 248 + // GetLastOp - /:did/log/last - latest op from audit log which isn't nullified 229 249 // if missing -> returns ErrDIDNotFound 230 250 // if tombstone -> returns tombstone op 231 251 plc.mu.Lock() ··· 245 265 return didplc.OpEnum{}, stacktrace.Propagate(ErrDIDNotFound, "") 246 266 } 247 267 248 248 - return l[len(l)-1].Operation, nil 268 268 + // find most recent operation that isn't nullified (during authoritative import, the latest operation might be nullified) 269 269 + for i := len(l) - 1; i >= 0; i-- { 270 270 + opEnum := l[i].Operation 271 271 + if !l[i].Nullified { 272 272 + return opEnum, nil 273 273 + } 274 274 + } 275 275 + // in the worst case all operations are somehow nullified and the loop ends with opEnum holding a nullified operation 276 276 + // that _shouldn't_ be possible (right?) but if it does happen, let's just behave as if the DID did not exist 277 277 + return didplc.OpEnum{}, stacktrace.Propagate(ErrDIDNotFound, "") 249 278 } 250 279 251 280 func (plc *plcImpl) Data(ctx context.Context, atHeight TreeVersion, did string) (didplc.RegularOp, error) { ··· 269 298 return didplc.RegularOp{}, stacktrace.Propagate(ErrDIDNotFound, "") 270 299 } 271 300 272 272 - opEnum := l[len(l)-1].Operation 273 273 - if opEnum.Tombstone != nil { 274 274 - return didplc.RegularOp{}, stacktrace.Propagate(ErrDIDGone, "") 301 301 + // find most recent operation that isn't nullified (during authoritative import, the latest operation might be nullified) 302 302 + for i := len(l) - 1; i >= 0; i-- { 303 303 + opEnum := l[i].Operation 304 304 + if !l[i].Nullified { 305 305 + if opEnum.Tombstone != nil { 306 306 + return didplc.RegularOp{}, stacktrace.Propagate(ErrDIDGone, "") 307 307 + } 308 308 + if opEnum.Regular != nil { 309 309 + return *opEnum.Regular, nil 310 310 + } 311 311 + return *modernizeOp(opEnum.Legacy), nil 312 312 + } 275 313 } 276 276 - if opEnum.Regular != nil { 277 277 - return *opEnum.Regular, nil 278 278 - } 279 279 - return *modernizeOp(opEnum.Legacy), nil 314 314 + // in the worst case all operations are somehow nullified and the loop ends with opEnum holding a nullified operation 315 315 + // that _shouldn't_ be possible (right?) but if it does happen, let's just behave as if the DID was tombstoned 316 316 + return didplc.RegularOp{}, stacktrace.Propagate(ErrDIDGone, "") 317 317 + 280 318 } 281 319 282 282 - func (plc *plcImpl) Export(ctx context.Context, atHeight TreeVersion, after time.Time, count int) ([]didplc.LogEntry, error) { 320 320 + func (plc *plcImpl) Export(ctx context.Context, atHeight TreeVersion, after uint64, count int) ([]types.SequencedLogEntry, error) { 283 321 plc.mu.Lock() 284 322 defer plc.mu.Unlock() 285 323 ··· 296 334 plc *plcImpl 297 335 } 298 336 299 299 - func (a *inMemoryAuditLogFetcher) AuditLogReverseIterator(atHeight TreeVersion, did string, retErr *error) iter.Seq2[int, didplc.LogEntry] { 337 337 + func (a *inMemoryAuditLogFetcher) AuditLogReverseIterator(atHeight TreeVersion, did string, retErr *error) iter.Seq2[int, types.SequencedLogEntry] { 300 338 tree, err := a.plc.treeProvider.ImmutableTree(atHeight) 301 339 if err != nil { 302 340 *retErr = stacktrace.Propagate(err, "") 303 303 - return func(yield func(int, didplc.LogEntry) bool) {} 341 341 + return func(yield func(int, types.SequencedLogEntry) bool) {} 304 342 } 305 343 306 344 return store.Tree.AuditLogReverseIterator(tree, did, retErr)

+9 -22

plc/operation_validator.go

··· 11 11 "github.com/did-method-plc/go-didplc" 12 12 "github.com/palantir/stacktrace" 13 13 "github.com/samber/mo" 14 14 + "tangled.org/gbl08ma/didplcbft/types" 14 15 ) 15 16 16 17 type AuditLogFetcher interface { 17 18 // AuditLogReverseIterator should return an iterator over the list of log entries for the specified DID, in reverse 18 18 - AuditLogReverseIterator(atHeight TreeVersion, did string, err *error) iter.Seq2[int, didplc.LogEntry] 19 19 + AuditLogReverseIterator(atHeight TreeVersion, did string, err *error) iter.Seq2[int, types.SequencedLogEntry] 19 20 } 20 21 21 22 type V0OperationValidator struct { ··· 74 75 75 76 proposedPrev := op.PrevCIDStr() 76 77 77 77 - partialLog := make(map[int]didplc.LogEntry) 78 78 + partialLog := make(map[int]types.SequencedLogEntry) 78 79 mostRecentOpIndex := -1 79 80 indexOfPrev := -1 80 81 var iteratorErr error ··· 88 89 } 89 90 } 90 91 91 91 - if entry.CID == proposedPrev { 92 92 + if entry.CID.String() == proposedPrev { 92 93 indexOfPrev = entryIdx 93 94 break 94 95 } ··· 98 99 return OperationEffects{}, stacktrace.Propagate(iteratorErr, "") 99 100 } 100 101 101 101 - nullifiedEntries := []didplc.LogEntry{} 102 102 + nullifiedEntries := []types.SequencedLogEntry{} 102 103 nullifiedEntriesStartingIndex := mo.None[int]() 103 104 104 105 if mostRecentOpIndex < 0 { ··· 125 126 126 127 // timestamps must increase monotonically 127 128 mostRecentOp := partialLog[mostRecentOpIndex] 128 128 - mostRecentCreatedAt, err := syntax.ParseDatetime(mostRecentOp.CreatedAt) 129 129 - if err != nil { 130 130 - return OperationEffects{}, stacktrace.Propagate(err, "reached invalid internal state") 131 131 - } 132 132 - if !timestamp.Time().After(mostRecentCreatedAt.Time()) { 129 129 + if !timestamp.Time().After(mostRecentOp.CreatedAt) { 133 130 return OperationEffects{}, stacktrace.Propagate(ErrInvalidOperationSequence, "") 134 131 } 135 132 ··· 156 153 } 157 154 158 155 // recovery key gets a 72hr window to do historical re-writes 159 159 - firstNullifiedCreatedAt, err := syntax.ParseDatetime(nullifiedEntries[0].CreatedAt) 160 160 - if err != nil { 161 161 - return OperationEffects{}, stacktrace.Propagate(err, "reached invalid internal state") 162 162 - } 163 163 - if timestamp.Time().Sub(firstNullifiedCreatedAt.Time()) > 72*time.Hour { 156 156 + if timestamp.Time().Sub(nullifiedEntries[0].CreatedAt) > 72*time.Hour { 164 157 return OperationEffects{}, stacktrace.Propagate(ErrRecoveryWindowExpired, "") 165 158 } 166 159 } else { ··· 230 223 for _, entry := range v.auditLogFetcher.AuditLogReverseIterator(atHeight, did, &err) { 231 224 if entry.Nullified { 232 225 // The typescript implementation operates over a `ops` array which doesn't include nullified ops 233 233 - // (With recovery ops also skipping rate limits, doesn't this leave the PLC vulnerable to the spam of constant recovery operations?) 226 226 + // (With recovery ops also skipping rate limits, doesn't this leave the PLC vulnerable to the spam of constant recovery operations? TODO investigate) 234 227 continue 235 228 } 236 236 - // Parse the CreatedAt timestamp string 237 237 - // The CreatedAt field is stored as a string in ISO 8601 format 238 238 - opDatetime, err := syntax.ParseDatetime(entry.CreatedAt) 239 239 - if err != nil { 240 240 - return stacktrace.Propagate(err, "") 241 241 - } 242 242 - opTime := opDatetime.Time() 229 229 + opTime := entry.CreatedAt 243 230 244 231 if opTime.Before(weekAgo) { 245 232 // operations are always ordered by timestamp, and we're iterating from newest to oldest

+2 -1

plc/plc.go

··· 7 7 8 8 "github.com/bluesky-social/indigo/atproto/syntax" 9 9 "github.com/did-method-plc/go-didplc" 10 10 + "tangled.org/gbl08ma/didplcbft/types" 10 11 ) 11 12 12 13 var ErrDIDNotFound = errors.New("DID not found") ··· 60 61 AuditLog(ctx context.Context, atHeight TreeVersion, did string) ([]didplc.LogEntry, error) 61 62 LastOperation(ctx context.Context, atHeight TreeVersion, did string) (didplc.OpEnum, error) 62 63 Data(ctx context.Context, atHeight TreeVersion, did string) (didplc.RegularOp, error) 63 63 - Export(ctx context.Context, atHeight TreeVersion, after time.Time, count int) ([]didplc.LogEntry, error) 64 64 + Export(ctx context.Context, atHeight TreeVersion, after uint64, count int) ([]types.SequencedLogEntry, error) 64 65 } 65 66 66 67 type WritePLC interface {

+132 -36

plc/plc_test.go

··· 16 16 "github.com/samber/lo" 17 17 "github.com/stretchr/testify/require" 18 18 "tangled.org/gbl08ma/didplcbft/plc" 19 19 + "tangled.org/gbl08ma/didplcbft/types" 19 20 ) 20 21 21 22 func TestPLC(t *testing.T) { ··· 190 191 doc, err = testPLC.Resolve(ctx, plc.SpecificTreeVersion(origVersion+4), testDID) 191 192 require.NoError(t, err) 192 193 193 193 - export, err := testPLC.Export(ctx, plc.CommittedTreeVersion, time.Time{}, 1000) 194 194 + export, err := testPLC.Export(ctx, plc.CommittedTreeVersion, 0, 1000) 194 195 require.NoError(t, err) 195 196 require.Len(t, export, 3) 196 197 197 198 require.Equal(t, "bafyreifgafcel2okxszhgbugieyvtmfig2gtf3dgqoh5fvdh3nlh6ncv6q", export[0].Operation.AsOperation().CID().String()) 198 198 - require.Equal(t, "bafyreifgafcel2okxszhgbugieyvtmfig2gtf3dgqoh5fvdh3nlh6ncv6q", export[0].CID) 199 199 + require.Equal(t, "bafyreifgafcel2okxszhgbugieyvtmfig2gtf3dgqoh5fvdh3nlh6ncv6q", export[0].CID.String()) 199 200 require.Equal(t, "bafyreia6ewwkwjgly6dijfepaq2ey6zximodbtqqi5f6fyugli3cxohn5m", export[1].Operation.AsOperation().CID().String()) 200 200 - require.Equal(t, "bafyreia6ewwkwjgly6dijfepaq2ey6zximodbtqqi5f6fyugli3cxohn5m", export[1].CID) 201 201 + require.Equal(t, "bafyreia6ewwkwjgly6dijfepaq2ey6zximodbtqqi5f6fyugli3cxohn5m", export[1].CID.String()) 201 202 require.Equal(t, "bafyreigyzl2esgnk7nvav5myvgywbshdmatzthc73iiar7tyeq3xjt47m4", export[2].Operation.AsOperation().CID().String()) 202 202 - require.Equal(t, "bafyreigyzl2esgnk7nvav5myvgywbshdmatzthc73iiar7tyeq3xjt47m4", export[2].CID) 203 203 + require.Equal(t, "bafyreigyzl2esgnk7nvav5myvgywbshdmatzthc73iiar7tyeq3xjt47m4", export[2].CID.String()) 203 204 204 204 - // the after parameter is exclusive, we should just get the second successful operation 205 205 - export, err = testPLC.Export(ctx, plc.CommittedTreeVersion, operations[1].ApplyAt.Time(), 1) 205 205 + // the after parameter is exclusive, with a limit of 1, we should just get the second successful operation 206 206 + export, err = testPLC.Export(ctx, plc.CommittedTreeVersion, export[0].Seq, 1) 206 207 require.NoError(t, err) 207 208 require.Len(t, export, 1) 208 208 - require.Equal(t, "bafyreia6ewwkwjgly6dijfepaq2ey6zximodbtqqi5f6fyugli3cxohn5m", export[0].CID) 209 209 + require.Equal(t, "bafyreia6ewwkwjgly6dijfepaq2ey6zximodbtqqi5f6fyugli3cxohn5m", export[0].CID.String()) 209 210 } 210 211 211 212 func TestPLCFromRemoteOperations(t *testing.T) { ··· 291 292 } 292 293 } 293 294 294 294 - export, err := testPLC.Export(ctx, plc.CommittedTreeVersion, time.Time{}, 0) 295 295 + export, err := testPLC.Export(ctx, plc.CommittedTreeVersion, 0, 0) 295 296 require.NoError(t, err) 296 296 - require.Len(t, export, 96) 297 297 + require.Len(t, export, 100) 297 298 298 299 // ensure entries are sorted correctly 299 299 - last := time.Time{} 300 300 + last := uint64(0) 300 301 for _, entry := range export { 301 301 - et, err := syntax.ParseDatetime(entry.CreatedAt) 302 302 - require.NoError(t, err) 303 303 - require.True(t, et.Time().After(last)) 304 304 - last = et.Time() 302 302 + require.True(t, entry.Seq > last) 303 303 + last = entry.Seq 305 304 } 306 305 } 307 306 ··· 412 411 require.NoError(t, err) 413 412 414 413 seenCIDs := map[string]struct{}{} 415 415 - for entry := range iterateOverExport(ctx, "") { 414 414 + for entry := range iterateOverExport(ctx, 0) { 416 415 err := testPLC.ImportOperationFromAuthoritativeSource(ctx, entry, func() ([]didplc.LogEntry, error) { 417 416 e, err := client.AuditLog(ctx, entry.DID) 418 417 return e, stacktrace.Propagate(err, "") ··· 420 419 require.NoError(t, err) 421 420 422 421 seenCIDs[entry.CID] = struct{}{} 423 423 - if len(seenCIDs) == 4000 { 422 422 + if len(seenCIDs) == 10000 { 424 423 break 425 424 } 426 425 } ··· 428 427 _, _, err = tree.SaveVersion() 429 428 require.NoError(t, err) 430 429 431 431 - exportedEntries, err := testPLC.Export(ctx, plc.CommittedTreeVersion, time.Time{}, len(seenCIDs)+1) 430 430 + exportedEntries, err := testPLC.Export(ctx, plc.CommittedTreeVersion, 0, len(seenCIDs)+1) 432 431 require.NoError(t, err) 433 432 434 433 require.Len(t, exportedEntries, len(seenCIDs)) 435 434 436 435 for _, exportedEntry := range exportedEntries { 437 437 - delete(seenCIDs, exportedEntry.CID) 436 436 + delete(seenCIDs, exportedEntry.CID.String()) 438 437 } 439 438 require.Empty(t, seenCIDs) 440 439 } 441 440 442 442 - func iterateOverExport(ctx context.Context, startAt string) iter.Seq[didplc.LogEntry] { 441 441 + func TestImportOperationWithNullification(t *testing.T) { 442 442 + var client didplc.Client 443 443 + 444 444 + ctx := t.Context() 445 445 + 446 446 + testFn := func(toImport []didplc.LogEntry, mutate func(didplc.LogEntry) didplc.LogEntry) ([]types.SequencedLogEntry, []didplc.LogEntry) { 447 447 + treeProvider := NewTestTreeProvider() 448 448 + testPLC := plc.NewPLC(treeProvider) 449 449 + 450 450 + tree, err := treeProvider.MutableTree() 451 451 + require.NoError(t, err) 452 452 + _, _, err = tree.SaveVersion() 453 453 + require.NoError(t, err) 454 454 + 455 455 + for _, entry := range toImport { 456 456 + entry = mutate(entry) 457 457 + err := testPLC.ImportOperationFromAuthoritativeSource(ctx, entry, func() ([]didplc.LogEntry, error) { 458 458 + e, err := client.AuditLog(ctx, entry.DID) 459 459 + return e, stacktrace.Propagate(err, "") 460 460 + }) 461 461 + require.NoError(t, err) 462 462 + } 463 463 + 464 464 + _, _, err = tree.SaveVersion() 465 465 + require.NoError(t, err) 466 466 + 467 467 + exportedEntries, err := testPLC.Export(ctx, plc.CommittedTreeVersion, 0, len(toImport)+1) 468 468 + require.NoError(t, err) 469 469 + 470 470 + require.Len(t, exportedEntries, len(toImport)) 471 471 + 472 472 + auditLog, err := testPLC.AuditLog(ctx, plc.CommittedTreeVersion, "did:plc:pkmfz5soq2swsvbhvjekb36g") 473 473 + require.NoError(t, err) 474 474 + 475 475 + return exportedEntries, auditLog 476 476 + } 477 477 + 478 478 + toImport, err := client.AuditLog(ctx, "did:plc:pkmfz5soq2swsvbhvjekb36g") 479 479 + require.NoError(t, err) 480 480 + 481 481 + exportedEntries, auditLog := testFn(toImport, func(le didplc.LogEntry) didplc.LogEntry { return le }) 482 482 + require.Len(t, auditLog, len(toImport)) 483 483 + 484 484 + for i, entry := range exportedEntries { 485 485 + require.Equal(t, uint64(i+1), entry.Seq) 486 486 + require.Equal(t, toImport[i].CID, entry.CID.String()) 487 487 + require.Equal(t, toImport[i].CID, auditLog[i].CID) 488 488 + require.Equal(t, toImport[i].CreatedAt, entry.CreatedAt.Format(types.ActualAtprotoDatetimeLayout)) 489 489 + require.Equal(t, toImport[i].CreatedAt, auditLog[i].CreatedAt) 490 490 + require.Equal(t, toImport[i].Nullified, entry.Nullified) 491 491 + require.Equal(t, toImport[i].Nullified, auditLog[i].Nullified) 492 492 + } 493 493 + 494 494 + // ensure auditLog never returns nullified entries as the last entries 495 495 + exportedEntries, auditLog = testFn(toImport[0:5], func(le didplc.LogEntry) didplc.LogEntry { return le }) 496 496 + 497 497 + require.Len(t, exportedEntries, 5) 498 498 + require.Len(t, auditLog, 1) 499 499 + require.False(t, auditLog[0].Nullified) 500 500 + require.Equal(t, auditLog[0].CID, "bafyreid2tbopmtuguvuvij5kjcqo7rv7yvqza37uvfcvk5zdxyo57xlfdi") 501 501 + 502 502 + // now pretend that at the time of import, no operations were nullified 503 503 + exportedEntries, auditLog = testFn(toImport, func(le didplc.LogEntry) didplc.LogEntry { 504 504 + le.Nullified = false 505 505 + return le 506 506 + }) 507 507 + require.Len(t, auditLog, len(toImport)) 508 508 + 509 509 + for i, entry := range exportedEntries { 510 510 + if i < 1 { 511 511 + require.Equal(t, uint64(i+1), entry.Seq) 512 512 + } else { 513 513 + require.Equal(t, uint64(i+5), entry.Seq) 514 514 + } 515 515 + require.Equal(t, toImport[i].CID, entry.CID.String()) 516 516 + require.Equal(t, toImport[i].CID, auditLog[i].CID) 517 517 + require.Equal(t, toImport[i].CreatedAt, entry.CreatedAt.Format(types.ActualAtprotoDatetimeLayout)) 518 518 + require.Equal(t, toImport[i].CreatedAt, auditLog[i].CreatedAt) 519 519 + require.Equal(t, toImport[i].Nullified, entry.Nullified) 520 520 + require.Equal(t, toImport[i].Nullified, auditLog[i].Nullified) 521 521 + } 522 522 + 523 523 + // now manipulate the timestamp on the first operation just to see the first operation get rewritten 524 524 + exportedEntries, auditLog = testFn(toImport, func(le didplc.LogEntry) didplc.LogEntry { 525 525 + if le.CID == "bafyreid2tbopmtuguvuvij5kjcqo7rv7yvqza37uvfcvk5zdxyo57xlfdi" { 526 526 + // this should cause mustFullyReplaceHistory to become true 527 527 + le.CreatedAt = syntax.DatetimeNow().String() 528 528 + } 529 529 + return le 530 530 + }) 531 531 + require.Len(t, auditLog, len(toImport)) 532 532 + 533 533 + for i, entry := range exportedEntries { 534 534 + require.Equal(t, uint64(i+2), entry.Seq) 535 535 + require.Equal(t, toImport[i].CID, entry.CID.String()) 536 536 + require.Equal(t, toImport[i].CID, auditLog[i].CID) 537 537 + require.Equal(t, toImport[i].CreatedAt, entry.CreatedAt.Format(types.ActualAtprotoDatetimeLayout)) 538 538 + require.Equal(t, toImport[i].CreatedAt, auditLog[i].CreatedAt) 539 539 + require.Equal(t, toImport[i].Nullified, entry.Nullified) 540 540 + require.Equal(t, toImport[i].Nullified, auditLog[i].Nullified) 541 541 + } 542 542 + } 543 543 + 544 544 + func iterateOverExport(ctx context.Context, startAt uint64) iter.Seq[didplc.LogEntry] { 443 545 return func(yield func(didplc.LogEntry) bool) { 444 546 const batchSize = 1000 445 547 baseURL := didplc.DefaultDirectoryURL + "/export" 446 548 client := &http.Client{Timeout: 30 * time.Second} 447 447 - 448 448 - // The /export seems to sometimes return outright duplicated entries :weary: 449 449 - seenCIDs := map[string]struct{}{} 450 549 451 550 after := startAt 452 551 for { ··· 459 558 460 559 q := req.URL.Query() 461 560 q.Add("count", fmt.Sprint(batchSize)) 462 462 - if after != "" { 463 463 - q.Add("after", after) 464 464 - } 561 561 + q.Add("after", fmt.Sprint(after)) 465 562 req.URL.RawQuery = q.Encode() 466 563 467 564 resp, err := client.Do(req) ··· 474 571 return // Non-200 status code 475 572 } 476 573 477 477 - entries := make([]didplc.LogEntry, 0, batchSize) 574 574 + type logEntryWithSeq struct { 575 575 + didplc.LogEntry 576 576 + Seq uint64 `json:"seq"` 577 577 + } 578 578 + 579 579 + entries := make([]logEntryWithSeq, 0, batchSize) 478 580 479 581 // Read response body 480 582 s := bufio.NewScanner(resp.Body) 481 583 receivedEntries := 0 482 584 for s.Scan() { 483 483 - var entry didplc.LogEntry 585 585 + var entry logEntryWithSeq 484 586 if err := json.Unmarshal(s.Bytes(), &entry); err != nil { 485 587 return // Failed to decode JSON 486 588 } 487 487 - if _, present := seenCIDs[entry.CID]; !present { 488 488 - entries = append(entries, entry) 489 489 - seenCIDs[entry.CID] = struct{}{} 490 490 - } 589 589 + entries = append(entries, entry) 491 590 receivedEntries++ 492 591 } 493 592 if s.Err() != nil { ··· 499 598 } 500 599 501 600 // Process each entry 502 502 - var lastCreatedAt string 503 601 for _, entry := range entries { 504 504 - lastCreatedAt = entry.CreatedAt 505 505 - if !yield(entry) { 602 602 + after = entry.Seq 603 603 + if !yield(entry.LogEntry) { 506 604 return 507 605 } 508 606 } ··· 510 608 if receivedEntries < batchSize { 511 609 return 512 610 } 513 513 - 514 514 - after = lastCreatedAt 515 611 } 516 612 } 517 613 }

+158 -108

store/tree.go

··· 4 4 "encoding/base32" 5 5 "encoding/binary" 6 6 "iter" 7 7 + "math" 7 8 "slices" 8 9 "strings" 9 10 "time" ··· 17 18 "github.com/polydawn/refmt/obj/atlas" 18 19 "github.com/samber/lo" 19 20 "github.com/samber/mo" 21 21 + "tangled.org/gbl08ma/didplcbft/types" 20 22 ) 21 23 22 22 - // ActualAtprotoDatetimeLayout is the format for CreatedAt timestamps 23 23 - // AtprotoDatetimeLayout as defined by github.com/bluesky-social/indigo/atproto/syntax omits trailing zeros in the milliseconds 24 24 - // This doesn't match how the official plc.directory implementation formats them, so we define that format here with trailing zeros included 25 25 - const ActualAtprotoDatetimeLayout = "2006-01-02T15:04:05.000Z" 26 26 - 27 24 var Tree PLCTreeStore = &TreeStore{} 28 25 29 26 type PLCTreeStore interface { 30 30 - AuditLog(tree ReadOnlyTree, did string, withProof bool) ([]didplc.LogEntry, *ics23.CommitmentProof, error) 31 31 - AuditLogReverseIterator(tree ReadOnlyTree, did string, err *error) iter.Seq2[int, didplc.LogEntry] 32 32 - ExportOperations(tree ReadOnlyTree, after time.Time, count int) ([]didplc.LogEntry, error) // passing a count of zero means unlimited 27 27 + AuditLog(tree ReadOnlyTree, did string, withProof bool) ([]types.SequencedLogEntry, *ics23.CommitmentProof, error) 28 28 + AuditLogReverseIterator(tree ReadOnlyTree, did string, err *error) iter.Seq2[int, types.SequencedLogEntry] 29 29 + ExportOperations(tree ReadOnlyTree, after uint64, count int) ([]types.SequencedLogEntry, error) // passing a count of zero means unlimited 33 30 StoreOperation(tree *iavl.MutableTree, entry didplc.LogEntry, nullifyWithIndexEqualOrGreaterThan mo.Option[int]) error 34 31 ReplaceHistory(tree *iavl.MutableTree, history []didplc.LogEntry) error 35 32 } ··· 39 36 // TreeStore exists just to groups methods nicely 40 37 type TreeStore struct{} 41 38 42 42 - func (t *TreeStore) AuditLog(tree ReadOnlyTree, did string, withProof bool) ([]didplc.LogEntry, *ics23.CommitmentProof, error) { 39 39 + func (t *TreeStore) AuditLog(tree ReadOnlyTree, did string, withProof bool) ([]types.SequencedLogEntry, *ics23.CommitmentProof, error) { 43 40 proofs := []*ics23.CommitmentProof{} 44 41 45 42 didBytes, err := didToBytes(did) ··· 61 58 return nil, nil, stacktrace.Propagate(err, "") 62 59 } 63 60 operationKeys = make([][]byte, 0, len(logOperations)/8) 64 64 - for ts := range slices.Chunk(logOperations, 8) { 65 65 - operationKeys = append(operationKeys, timestampBytesToDIDOperationKey(ts, didBytes)) 61 61 + for seqBytes := range slices.Chunk(logOperations, 8) { 62 62 + operationKeys = append(operationKeys, sequenceBytesToOperationKey(seqBytes)) 66 63 } 67 64 } 68 65 ··· 74 71 proofs = append(proofs, proof) 75 72 } 76 73 77 77 - logEntries := make([]didplc.LogEntry, 0, len(operationKeys)) 74 74 + logEntries := make([]types.SequencedLogEntry, 0, len(operationKeys)) 78 75 for _, opKey := range operationKeys { 79 76 operationValue, err := tree.Get(opKey) 80 77 if err != nil { ··· 89 86 proofs = append(proofs, proof) 90 87 } 91 88 92 92 - nullified, operation, err := unmarshalOperationValue(operationValue) 89 89 + logEntry, err := unmarshalLogEntry(opKey, operationValue) 93 90 if err != nil { 94 91 return nil, nil, stacktrace.Propagate(err, "") 95 92 } 96 93 97 97 - timestamp, actualDID, err := unmarshalOperationKey(opKey) 98 98 - if err != nil { 99 99 - return nil, nil, stacktrace.Propagate(err, "") 100 100 - } 101 101 - 102 102 - logEntries = append(logEntries, didplc.LogEntry{ 103 103 - DID: actualDID, 104 104 - Operation: operation, 105 105 - CID: operation.AsOperation().CID().String(), 106 106 - Nullified: nullified, 107 107 - CreatedAt: timestamp.Format(ActualAtprotoDatetimeLayout), 108 108 - }) 94 94 + logEntries = append(logEntries, logEntry) 109 95 } 110 96 111 97 var combinedProof *ics23.CommitmentProof ··· 118 104 return logEntries, combinedProof, nil 119 105 } 120 106 121 121 - func (t *TreeStore) AuditLogReverseIterator(tree ReadOnlyTree, did string, retErr *error) iter.Seq2[int, didplc.LogEntry] { 122 122 - return func(yield func(int, didplc.LogEntry) bool) { 107 107 + func (t *TreeStore) AuditLogReverseIterator(tree ReadOnlyTree, did string, retErr *error) iter.Seq2[int, types.SequencedLogEntry] { 108 108 + return func(yield func(int, types.SequencedLogEntry) bool) { 123 109 didBytes, err := didToBytes(did) 124 110 if err != nil { 125 111 *retErr = stacktrace.Propagate(err, "") ··· 142 128 return 143 129 } 144 130 operationKeys = make([][]byte, 0, len(logOperations)/8) 145 145 - for ts := range slices.Chunk(logOperations, 8) { 146 146 - operationKeys = append(operationKeys, timestampBytesToDIDOperationKey(ts, didBytes)) 131 131 + for seqBytes := range slices.Chunk(logOperations, 8) { 132 132 + operationKeys = append(operationKeys, sequenceBytesToOperationKey(seqBytes)) 147 133 } 148 134 } 149 135 ··· 155 141 return 156 142 } 157 143 158 158 - nullified, operation, err := unmarshalOperationValue(operationValue) 144 144 + logEntry, err := unmarshalLogEntry(opKey, operationValue) 159 145 if err != nil { 160 146 *retErr = stacktrace.Propagate(err, "") 161 147 return 162 148 } 163 149 164 164 - timestamp, actualDID, err := unmarshalOperationKey(opKey) 165 165 - if err != nil { 166 166 - *retErr = stacktrace.Propagate(err, "") 167 167 - return 168 168 - } 169 169 - 170 170 - if !yield(i, didplc.LogEntry{ 171 171 - DID: actualDID, 172 172 - Operation: operation, 173 173 - CID: operation.AsOperation().CID().String(), 174 174 - Nullified: nullified, 175 175 - CreatedAt: timestamp.Format(ActualAtprotoDatetimeLayout), 176 176 - }) { 150 150 + if !yield(i, logEntry) { 177 151 return 178 152 } 179 153 } 180 154 } 181 155 } 182 156 183 183 - func (t *TreeStore) ExportOperations(tree ReadOnlyTree, after time.Time, count int) ([]didplc.LogEntry, error) { 157 157 + func (t *TreeStore) ExportOperations(tree ReadOnlyTree, after uint64, count int) ([]types.SequencedLogEntry, error) { 184 158 // as the name suggests, after is an exclusive lower bound, but our iterators use inclusive lower bounds 185 185 - start := after.Add(1 * time.Nanosecond) 186 186 - startKey := marshalOperationKey(start, make([]byte, 15)) 187 187 - if after.UnixNano() < 0 { 188 188 - // our storage format doesn't deal well with negative unix timestamps, 189 189 - // but that's fine because we don't have operations created that far back. assume we just want to iterate from the start 190 190 - copy(startKey[1:8], make([]byte, 8)) 191 191 - } 159 159 + start := after + 1 160 160 + startKey := marshalOperationKey(start) 161 161 + endKey := maxOperationKey 192 162 193 193 - entries := make([]didplc.LogEntry, 0, count) 163 163 + entries := make([]types.SequencedLogEntry, 0, count) 194 164 var iterErr error 195 195 - tree.IterateRange(startKey, nil, true, func(operationKey, operationValue []byte) bool { 196 196 - nullified, operation, err := unmarshalOperationValue(operationValue) 165 165 + tree.IterateRange(startKey, endKey, true, func(operationKey, operationValue []byte) bool { 166 166 + logEntry, err := unmarshalLogEntry(operationKey, operationValue) 197 167 if err != nil { 198 168 iterErr = stacktrace.Propagate(err, "") 199 169 return true 200 170 } 201 171 202 202 - timestamp, actualDID, err := unmarshalOperationKey(operationKey) 203 203 - if err != nil { 204 204 - iterErr = stacktrace.Propagate(err, "") 205 205 - return true 206 206 - } 207 207 - 208 208 - entries = append(entries, didplc.LogEntry{ 209 209 - DID: actualDID, 210 210 - Operation: operation, 211 211 - CID: operation.AsOperation().CID().String(), 212 212 - Nullified: nullified, 213 213 - CreatedAt: timestamp.Format(ActualAtprotoDatetimeLayout), 214 214 - }) 172 172 + entries = append(entries, logEntry) 215 173 return len(entries) == count // this condition being checked here also makes it so that a count of zero means unlimited 216 174 }) 217 175 if iterErr != nil { ··· 237 195 operationKeys = [][]byte{} 238 196 } else { 239 197 operationKeys = make([][]byte, 0, len(logOperations)/8) 240 240 - for ts := range slices.Chunk(logOperations, 8) { 241 241 - operationKeys = append(operationKeys, timestampBytesToDIDOperationKey(ts, didBytes)) 198 198 + for seqBytes := range slices.Chunk(logOperations, 8) { 199 199 + operationKeys = append(operationKeys, sequenceBytesToOperationKey(seqBytes)) 242 200 } 243 201 } 244 202 ··· 262 220 return stacktrace.Propagate(err, "invalid CreatedAt") 263 221 } 264 222 223 223 + seq, err := getNextSeqID(tree) 224 224 + if err != nil { 225 225 + return stacktrace.Propagate(err, "") 226 226 + } 227 227 + 265 228 operation := entry.Operation.AsOperation() 266 266 - opKey := marshalOperationKey(opDatetime.Time(), didBytes) 267 267 - opValue := marshalOperationValue(entry.Nullified, operation) 229 229 + opKey := marshalOperationKey(seq) 230 230 + opValue := marshalOperationValue(entry.Nullified, didBytes, opDatetime.Time(), operation) 268 231 269 232 _, err = tree.Set(opKey, opValue) 270 233 if err != nil { ··· 280 243 return nil 281 244 } 282 245 283 283 - func (t *TreeStore) ReplaceHistory(tree *iavl.MutableTree, history []didplc.LogEntry) error { 284 284 - if len(history) == 0 { 246 246 + func (t *TreeStore) ReplaceHistory(tree *iavl.MutableTree, remoteHistory []didplc.LogEntry) error { 247 247 + if len(remoteHistory) == 0 { 285 248 // for now this isn't needed, if it's needed in the future we'll have to accept a DID as argument on this function 286 249 return stacktrace.NewError("can't replace with empty history") 287 250 } 288 251 289 289 - did := history[0].DID 252 252 + did := remoteHistory[0].DID 290 253 291 254 didBytes, err := didToBytes(did) 292 255 if err != nil { ··· 295 258 296 259 logKey := marshalDIDLogKey(didBytes) 297 260 298 298 - // identify keys of existing operations for this DID (if any) 299 299 - var prevOpKeys [][]byte 300 300 - logOperations, err := tree.Get(logKey) 261 261 + localHistory, _, err := t.AuditLog(tree, did, false) 301 262 if err != nil { 302 263 return stacktrace.Propagate(err, "") 303 264 } 304 304 - prevOpKeys = make([][]byte, 0, len(logOperations)/8) 305 305 - for ts := range slices.Chunk(logOperations, 8) { 306 306 - prevOpKeys = append(prevOpKeys, timestampBytesToDIDOperationKey(ts, didBytes)) 265 265 + 266 266 + // if the first operations are equal to what we already have, keep them untouched to minimize the turmoil 267 267 + keepLocalBeforeIdx := 0 268 268 + for i, localEntry := range localHistory { 269 269 + if i >= len(remoteHistory) { 270 270 + break 271 271 + } 272 272 + remoteEntry := remoteHistory[i] 273 273 + 274 274 + // stop looping once we find a difference 275 275 + // we trust that the authoritative source computes CIDs properly (i.e. that two operations having the same CID are indeed equal) 276 276 + if localEntry.Nullified != remoteEntry.Nullified || localEntry.CID.String() != remoteEntry.CID { 277 277 + break 278 278 + } 279 279 + 280 280 + remoteDatetime, err := syntax.ParseDatetime(remoteEntry.CreatedAt) 281 281 + if err != nil { 282 282 + return stacktrace.Propagate(err, "invalid CreatedAt") 283 283 + } 284 284 + 285 285 + if !localEntry.CreatedAt.Equal(remoteDatetime.Time()) { 286 286 + break 287 287 + } 288 288 + 289 289 + keepLocalBeforeIdx++ 307 290 } 308 291 309 309 - // remove existing operations for this DID (if any) 310 310 - for _, key := range prevOpKeys { 292 292 + // all replaced/added operations get new sequence IDs. 293 293 + // Get the highest sequence ID before removing any keys to ensure the sequence IDs actually change 294 294 + seq, err := getNextSeqID(tree) 295 295 + if err != nil { 296 296 + return stacktrace.Propagate(err, "") 297 297 + } 298 298 + 299 299 + // remove existing conflicting operations for this DID (if any) 300 300 + logOperations, err := tree.Get(logKey) 301 301 + if err != nil { 302 302 + return stacktrace.Propagate(err, "") 303 303 + } 304 304 + logOperationsToDelete := logOperations[8*keepLocalBeforeIdx:] 305 305 + for seqBytes := range slices.Chunk(logOperationsToDelete, 8) { 306 306 + key := sequenceBytesToOperationKey(seqBytes) 307 307 + 311 308 _, _, err = tree.Remove(key) 312 309 if err != nil { 313 310 return stacktrace.Propagate(err, "") 314 311 } 315 312 } 316 313 317 317 - // add new list of operations 318 318 - logOperations = make([]byte, 0, len(history)*8) 319 319 - for _, entry := range history { 314 314 + // add just the operations past the point they weren't kept 315 315 + remoteHistory = remoteHistory[keepLocalBeforeIdx:] 316 316 + 317 317 + // keep the operations log up until the point we've kept the history 318 318 + // clone just to make sure we avoid issues since we got this slice from the tree, it is not meant to be modified 319 319 + logOperations = slices.Clone(logOperations[0 : 8*keepLocalBeforeIdx]) 320 320 + 321 321 + for _, entry := range remoteHistory { 320 322 opDatetime, err := syntax.ParseDatetime(entry.CreatedAt) 321 323 if err != nil { 322 324 return stacktrace.Propagate(err, "invalid CreatedAt") 323 325 } 324 326 325 327 operation := entry.Operation.AsOperation() 326 326 - opKey := marshalOperationKey(opDatetime.Time(), didBytes) 327 327 - opValue := marshalOperationValue(entry.Nullified, operation) 328 328 + opKey := marshalOperationKey(seq) 329 329 + seq++ 330 330 + opValue := marshalOperationValue(entry.Nullified, didBytes, opDatetime.Time(), operation) 328 331 329 332 _, err = tree.Set(opKey, opValue) 330 333 if err != nil { ··· 344 347 return nil 345 348 } 346 349 350 350 + var minOperationKey = marshalOperationKey(0) 351 351 + var maxOperationKey = marshalOperationKey(math.MaxInt64) 352 352 + 353 353 + func getNextSeqID(tree *iavl.MutableTree) (uint64, error) { 354 354 + seq := uint64(0) 355 355 + var err error 356 356 + tree.IterateRange(minOperationKey, maxOperationKey, false, func(key, value []byte) bool { 357 357 + seq, err = unmarshalOperationKey(key) 358 358 + return true 359 359 + }) 360 360 + 361 361 + return seq + 1, stacktrace.Propagate(err, "") 362 362 + } 363 363 + 347 364 func didToBytes(did string) ([]byte, error) { 348 365 if !strings.HasPrefix(did, "did:plc:") { 349 366 return nil, stacktrace.NewError("invalid did:plc") ··· 379 396 return key 380 397 } 381 398 382 382 - func timestampBytesToDIDOperationKey(timestamp []byte, didBytes []byte) []byte { 383 383 - key := make([]byte, 1+8+15) 399 399 + func sequenceBytesToOperationKey(sequenceBytes []byte) []byte { 400 400 + key := make([]byte, 1+8) 384 401 key[0] = 'o' 385 385 - copy(key[1:9], timestamp) 386 386 - copy(key[9:], didBytes) 402 402 + copy(key[1:9], sequenceBytes) 387 403 return key 388 404 } 389 405 390 390 - func marshalOperationKey(createdAt time.Time, didBytes []byte) []byte { 391 391 - key := make([]byte, 1+8+15) 406 406 + func marshalOperationKey(sequence uint64) []byte { 407 407 + key := make([]byte, 1+8) 392 408 key[0] = 'o' 393 409 394 394 - ts := uint64(createdAt.Truncate(1 * time.Millisecond).UTC().UnixNano()) 395 395 - binary.BigEndian.PutUint64(key[1:], ts) 410 410 + binary.BigEndian.PutUint64(key[1:], sequence) 396 411 397 397 - copy(key[9:], didBytes) 398 412 return key 399 413 } 400 414 401 401 - func unmarshalOperationKey(key []byte) (time.Time, string, error) { 402 402 - createdAtUnixNano := binary.BigEndian.Uint64(key[1:9]) 403 403 - createdAt := time.Unix(0, int64(createdAtUnixNano)).UTC() 404 404 - did, err := bytesToDID(key[9:]) 405 405 - return createdAt, did, stacktrace.Propagate(err, "") 415 415 + func unmarshalOperationKey(key []byte) (uint64, error) { 416 416 + return binary.BigEndian.Uint64(key[1:9]), nil 406 417 } 407 418 408 408 - func marshalOperationValue(nullified bool, operation didplc.Operation) []byte { 409 409 - o := []byte{lo.Ternary[byte](nullified, 1, 0)} 410 410 - o = append(o, operation.SignedCBORBytes()...) 419 419 + func marshalOperationValue(nullified bool, didBytes []byte, createdAt time.Time, operation didplc.Operation) []byte { 420 420 + opAsBytes := operation.SignedCBORBytes() 421 421 + o := make([]byte, 1+15+8+len(opAsBytes)) 422 422 + 423 423 + o[0] = lo.Ternary[byte](nullified, 1, 0) 424 424 + 425 425 + copy(o[1:16], didBytes) 426 426 + 427 427 + ts := uint64(createdAt.Truncate(1 * time.Millisecond).UTC().UnixNano()) 428 428 + binary.BigEndian.PutUint64(o[16:24], ts) 429 429 + copy(o[24:], opAsBytes) 430 430 + 411 431 return o 412 432 } 413 433 414 414 - func unmarshalOperationValue(value []byte) (bool, didplc.OpEnum, error) { 434 434 + func unmarshalOperationValue(value []byte) (bool, string, time.Time, didplc.OpEnum, error) { 415 435 nullified := value[0] != 0 436 436 + 437 437 + did, err := bytesToDID(value[1:16]) 438 438 + if err != nil { 439 439 + return false, "", time.Time{}, didplc.OpEnum{}, stacktrace.Propagate(err, "") 440 440 + } 441 441 + 442 442 + createdAtUnixNano := binary.BigEndian.Uint64(value[16:24]) 443 443 + createdAt := time.Unix(0, int64(createdAtUnixNano)).UTC() 444 444 + 416 445 var opEnum didplc.OpEnum 417 417 - err := cbornode.DecodeInto(value[1:], &opEnum) 446 446 + err = cbornode.DecodeInto(value[24:], &opEnum) 447 447 + if err != nil { 448 448 + return false, "", time.Time{}, didplc.OpEnum{}, stacktrace.Propagate(err, "") 449 449 + } 450 450 + return nullified, did, createdAt, opEnum, nil 451 451 + } 452 452 + 453 453 + func unmarshalLogEntry(operationKey, operationValue []byte) (types.SequencedLogEntry, error) { 454 454 + nullified, actualDID, timestamp, operation, err := unmarshalOperationValue(operationValue) 455 455 + if err != nil { 456 456 + return types.SequencedLogEntry{}, stacktrace.Propagate(err, "") 457 457 + } 458 458 + 459 459 + seq, err := unmarshalOperationKey(operationKey) 418 460 if err != nil { 419 419 - return false, didplc.OpEnum{}, stacktrace.Propagate(err, "") 461 461 + return types.SequencedLogEntry{}, stacktrace.Propagate(err, "") 420 462 } 421 421 - return nullified, opEnum, nil 463 463 + 464 464 + return types.SequencedLogEntry{ 465 465 + Seq: seq, 466 466 + DID: actualDID, 467 467 + Operation: operation, 468 468 + CID: operation.AsOperation().CID(), 469 469 + Nullified: nullified, 470 470 + CreatedAt: timestamp, 471 471 + }, nil 422 472 } 423 473 424 474 func init() {

+32

types/log_entry.go

··· 1 1 + package types 2 2 + 3 3 + import ( 4 4 + "time" 5 5 + 6 6 + "github.com/did-method-plc/go-didplc" 7 7 + "github.com/ipfs/go-cid" 8 8 + ) 9 9 + 10 10 + type SequencedLogEntry struct { 11 11 + Seq uint64 12 12 + DID string 13 13 + Operation didplc.OpEnum 14 14 + CID cid.Cid 15 15 + Nullified bool 16 16 + CreatedAt time.Time 17 17 + } 18 18 + 19 19 + func (l SequencedLogEntry) ToDIDPLCLogEntry() didplc.LogEntry { 20 20 + return didplc.LogEntry{ 21 21 + DID: l.DID, 22 22 + Operation: l.Operation, 23 23 + CID: l.CID.String(), 24 24 + Nullified: l.Nullified, 25 25 + CreatedAt: l.CreatedAt.Format(ActualAtprotoDatetimeLayout), 26 26 + } 27 27 + } 28 28 + 29 29 + // ActualAtprotoDatetimeLayout is the format for CreatedAt timestamps 30 30 + // AtprotoDatetimeLayout as defined by github.com/bluesky-social/indigo/atproto/syntax omits trailing zeros in the milliseconds 31 31 + // This doesn't match how the official plc.directory implementation formats them, so we define that format here with trailing zeros included 32 32 + const ActualAtprotoDatetimeLayout = "2006-01-02T15:04:05.000Z"