alice.mosphere.at / indigo
this repo has no description
0
fork

Configure Feed

Select the types of activity you want to include in your feed.

add support for handling rebase events (#175)

I still want to add more tests to this, but this is done enough to start
in on some code review.

authored by

Whyrusleeping and committed by
GitHub 8be05e41 a7315631

+502 -43
+9 -1
bgs/bgs.go
··· 431 431 return nil 432 432 } 433 433 434 - // TODO: if the user is already in the 'slow' path, we shouldnt even bother trying to fast path this event 434 + // skip the fast path for rebases or if the user is already in the slow path 435 + if evt.Rebase || bgs.Index.Crawler.RepoInSlowPath(ctx, host, u.ID) { 436 + ai, err := bgs.Index.LookupUser(ctx, u.ID) 437 + if err != nil { 438 + return err 439 + } 440 + 441 + return bgs.Index.Crawler.AddToCatchupQueue(ctx, host, ai, evt) 442 + } 435 443 436 444 if err := bgs.repoman.HandleExternalUserEvent(ctx, host.ID, u.ID, u.Did, (*cid.Cid)(evt.Prev), evt.Blocks); err != nil { 437 445 log.Warnw("failed handling event", "err", err, "host", host.Host, "seq", evt.Seq, "repo", u.Did, "prev", stringLink(evt.Prev), "commit", evt.Commit.String())
+90 -14
carstore/bs.go
··· 70 70 Seq int `gorm:"index"` 71 71 Path string 72 72 Usr util.Uid `gorm:"index"` 73 + Rebase bool 73 74 } 74 75 75 76 type blockRef struct { ··· 344 345 if earlyCid.Defined() { 345 346 var untilShard CarShard 346 347 if err := cs.meta.First(&untilShard, "root = ? AND usr = ?", util.DbCID{earlyCid}, user).Error; err != nil { 347 - return err 348 + return fmt.Errorf("finding early shard: %w", err) 348 349 } 349 350 earlySeq = untilShard.Seq 350 351 } ··· 352 353 if lateCid.Defined() { 353 354 var fromShard CarShard 354 355 if err := cs.meta.First(&fromShard, "root = ? AND usr = ?", util.DbCID{lateCid}, user).Error; err != nil { 355 - return err 356 + return fmt.Errorf("finding late shard: %w", err) 356 357 } 357 358 lateSeq = fromShard.Seq 358 359 } ··· 384 385 } 385 386 386 387 for _, sh := range shards { 387 - if err := cs.writeShardBlocks(ctx, &sh, w); err != nil { 388 - return err 388 + // for rebase shards, only include the modified root, not the whole tree 389 + if sh.Rebase && incremental { 390 + if err := cs.writeBlockFromShard(ctx, &sh, w, sh.Root.CID); err != nil { 391 + return err 392 + } 393 + } else { 394 + if err := cs.writeShardBlocks(ctx, &sh, w); err != nil { 395 + return err 396 + } 389 397 } 390 398 } 391 399 ··· 415 423 return nil 416 424 } 417 425 426 + func (cs *CarStore) writeBlockFromShard(ctx context.Context, sh *CarShard, w io.Writer, c cid.Cid) error { 427 + fi, err := os.Open(sh.Path) 428 + if err != nil { 429 + return err 430 + } 431 + defer fi.Close() 432 + 433 + rr, err := car.NewCarReader(fi) 434 + if err != nil { 435 + return err 436 + } 437 + 438 + for { 439 + blk, err := rr.Next() 440 + if err != nil { 441 + return err 442 + } 443 + 444 + if blk.Cid() == c { 445 + _, err := LdWrite(w, c.Bytes(), blk.RawData()) 446 + return err 447 + } 448 + } 449 + } 450 + 418 451 var _ blockstore.Blockstore = (*DeltaSession)(nil) 419 452 420 453 func (ds *DeltaSession) Put(ctx context.Context, b blockformat.Block) error { ··· 515 548 // CloseWithRoot writes all new blocks in a car file to the writer with the 516 549 // given cid as the 'root' 517 550 func (ds *DeltaSession) CloseWithRoot(ctx context.Context, root cid.Cid) ([]byte, error) { 518 - ctx, span := otel.Tracer("carstore").Start(ctx, "CloseWithRoot") 519 - defer span.End() 551 + return ds.closeWithRoot(ctx, root, false) 552 + } 520 553 521 - if ds.readonly { 522 - return nil, fmt.Errorf("cannot write to readonly deltaSession") 523 - } 524 - 525 - buf := new(bytes.Buffer) 554 + func WriteCarHeader(w io.Writer, root cid.Cid) (int64, error) { 526 555 h := &car.CarHeader{ 527 556 Roots: []cid.Cid{root}, 528 557 Version: 1, 529 558 } 530 559 hb, err := cbor.DumpObject(h) 531 560 if err != nil { 532 - return nil, err 561 + return 0, err 533 562 } 534 563 535 - hnw, err := LdWrite(buf, hb) 564 + hnw, err := LdWrite(w, hb) 565 + if err != nil { 566 + return 0, err 567 + } 568 + 569 + return hnw, nil 570 + } 571 + 572 + func (ds *DeltaSession) closeWithRoot(ctx context.Context, root cid.Cid, rebase bool) ([]byte, error) { 573 + ctx, span := otel.Tracer("carstore").Start(ctx, "CloseWithRoot") 574 + defer span.End() 575 + 576 + if ds.readonly { 577 + return nil, fmt.Errorf("cannot write to readonly deltaSession") 578 + } 579 + 580 + buf := new(bytes.Buffer) 581 + hnw, err := WriteCarHeader(buf, root) 536 582 if err != nil { 537 583 return nil, err 538 584 } ··· 540 586 // TODO: writing these blocks in map traversal order is bad, I believe the 541 587 // optimal ordering will be something like reverse-write-order, but random 542 588 // is definitely not it 543 - var offset int64 = hnw 589 + 590 + offset := hnw 544 591 //brefs := make([]*blockRef, 0, len(ds.blks)) 545 592 brefs := make([]map[string]interface{}, 0, len(ds.blks)) 546 593 for k, blk := range ds.blks { ··· 602 649 } 603 650 604 651 return buf.Bytes(), nil 652 + } 653 + 654 + func (ds *DeltaSession) CloseAsRebase(ctx context.Context, root cid.Cid) error { 655 + _, err := ds.closeWithRoot(ctx, root, true) 656 + if err != nil { 657 + return err 658 + } 659 + 660 + // TODO: this *could* get large, might be worth doing it incrementally 661 + var oldslices []CarShard 662 + if err := ds.cs.meta.Find(&oldslices, "usr = ? AND seq < ?", ds.user, ds.seq).Error; err != nil { 663 + return err 664 + } 665 + 666 + // If anything here fails, cleanup is straightforward. Simply look for any 667 + // shard in the database with a higher seq shard marked as 'rebase' 668 + for _, sl := range oldslices { 669 + if err := os.Remove(sl.Path); err != nil { 670 + if !os.IsNotExist(err) { 671 + return err 672 + } 673 + } 674 + 675 + if err := ds.cs.meta.Delete(&sl).Error; err != nil { 676 + return err 677 + } 678 + } 679 + 680 + return nil 605 681 } 606 682 607 683 func LdWrite(w io.Writer, d ...[]byte) (int64, error) {
+18 -6
events/dbpersist.go
··· 28 28 Commit util.DbCID 29 29 Prev *util.DbCID 30 30 31 - Time time.Time 32 - Blobs []byte 33 - Repo util.Uid 34 - Event string 31 + Time time.Time 32 + Blobs []byte 33 + Repo util.Uid 34 + Event string 35 + Rebase bool 35 36 36 37 Ops []RepoOpRecord 37 38 } ··· 103 104 Event: "repo_append", // TODO: refactor to "#commit"? can "rebase" come through this path? 104 105 Blobs: blobs, 105 106 Time: t, 107 + Rebase: evt.Rebase, 106 108 } 107 109 108 110 for _, op := range evt.Ops { ··· 210 212 Prev: prevCID, 211 213 Time: rer.Time.Format(util.ISO8601), 212 214 Blobs: blobCIDs, 215 + Rebase: rer.Rebase, 213 216 // TODO: there was previously an Event field here. are these all Commit, or are some other events? 214 217 } 215 218 ··· 229 232 230 233 cs, err := p.readCarSlice(ctx, rer) 231 234 if err != nil { 232 - return nil, fmt.Errorf("read car slice: %w", err) 235 + return nil, fmt.Errorf("read car slice (%s): %w", rer.Commit.CID, err) 233 236 } 234 237 235 238 if len(cs) > carstore.MaxSliceLength { ··· 244 247 func (p *DbPersistence) readCarSlice(ctx context.Context, rer *RepoEventRecord) ([]byte, error) { 245 248 246 249 var early cid.Cid 247 - if rer.Prev != nil { 250 + if rer.Prev != nil && !rer.Rebase { 248 251 early = rer.Prev.CID 249 252 } 250 253 ··· 257 260 } 258 261 259 262 func (p *DbPersistence) TakeDownRepo(ctx context.Context, usr util.Uid) error { 263 + return p.deleteAllEventsForUser(ctx, usr) 264 + } 265 + 266 + func (p *DbPersistence) deleteAllEventsForUser(ctx context.Context, usr util.Uid) error { 260 267 for { 261 268 q := p.db.Model(&RepoEventRecord{}).Where("repo = ?", usr).Limit(100).Select("seq") 262 269 res := p.db.Where("repo_event_record_id in (?)", q).Delete(&RepoOpRecord{}) ··· 275 282 276 283 return nil 277 284 } 285 + 286 + func (p *DbPersistence) RebaseRepoEvents(ctx context.Context, usr util.Uid) error { 287 + // a little weird that this is the same action as a takedown 288 + return p.deleteAllEventsForUser(ctx, usr) 289 + }
+4
events/events.go
··· 221 221 func (em *EventManager) TakeDownRepo(ctx context.Context, user util.Uid) error { 222 222 return em.persister.TakeDownRepo(ctx, user) 223 223 } 224 + 225 + func (em *EventManager) HandleRebase(ctx context.Context, user util.Uid) error { 226 + return em.persister.RebaseRepoEvents(ctx, user) 227 + }
+5
events/persist.go
··· 13 13 Persist(ctx context.Context, e *XRPCStreamEvent) error 14 14 Playback(ctx context.Context, since int64, cb func(*XRPCStreamEvent) error) error 15 15 TakeDownRepo(ctx context.Context, usr util.Uid) error 16 + RebaseRepoEvents(ctx context.Context, usr util.Uid) error 16 17 } 17 18 18 19 // MemPersister is the most naive implementation of event persistence ··· 73 74 func (mp *MemPersister) TakeDownRepo(ctx context.Context, uid util.Uid) error { 74 75 return fmt.Errorf("repo takedowns not currently supported by memory persister, test usage only") 75 76 } 77 + 78 + func (mp *MemPersister) RebaseRepoEvents(ctx context.Context, usr util.Uid) error { 79 + return fmt.Errorf("repo rebases not currently supported by memory persister, test usage only") 80 + }
+48 -14
indexer/crawler.go
··· 3 3 import ( 4 4 "context" 5 5 "fmt" 6 + "sync" 6 7 7 8 comatproto "github.com/bluesky-social/indigo/api/atproto" 8 9 "github.com/bluesky-social/indigo/models" ··· 20 21 21 22 complete chan util.Uid 22 23 24 + maplk sync.Mutex 25 + todo map[util.Uid]*crawlWork 26 + inProgress map[util.Uid]*crawlWork 27 + 23 28 doRepoCrawl func(context.Context, *crawlWork) error 24 29 25 30 concurrency int ··· 37 42 catchup: make(chan *catchupJob), 38 43 doRepoCrawl: repoFn, 39 44 concurrency: concurrency, 45 + todo: make(map[util.Uid]*crawlWork), 46 + inProgress: make(map[util.Uid]*crawlWork), 40 47 }, nil 41 48 } 42 49 ··· 62 69 63 70 // for events that come in while this actor is being processed 64 71 next []*catchupJob 72 + 73 + rebase *catchupJob 65 74 } 66 75 67 76 func (c *CrawlDispatcher) mainLoop() { 68 77 var next *crawlWork 69 78 var buffer []*crawlWork 70 - 71 - todo := make(map[util.Uid]*crawlWork) 72 - inProgress := make(map[util.Uid]*crawlWork) 73 79 74 80 var rs chan *crawlWork 75 81 for { ··· 77 83 case act := <-c.ingest: 78 84 // TODO: max buffer size 79 85 80 - _, ok := inProgress[act.Uid] 86 + c.maplk.Lock() 87 + _, ok := c.inProgress[act.Uid] 81 88 if ok { 89 + c.maplk.Unlock() 82 90 break 83 91 } 84 92 85 - _, has := todo[act.Uid] 93 + _, has := c.todo[act.Uid] 86 94 if has { 95 + c.maplk.Unlock() 87 96 break 88 97 } 89 98 ··· 91 100 act: act, 92 101 initScrape: true, 93 102 } 94 - todo[act.Uid] = cw 103 + c.todo[act.Uid] = cw 104 + c.maplk.Unlock() 95 105 96 106 if next == nil { 97 107 next = cw ··· 100 110 buffer = append(buffer, cw) 101 111 } 102 112 case rs <- next: 103 - delete(todo, next.act.Uid) 104 - inProgress[next.act.Uid] = next 113 + c.maplk.Lock() 114 + delete(c.todo, next.act.Uid) 115 + c.inProgress[next.act.Uid] = next 116 + c.maplk.Unlock() 105 117 106 118 if len(buffer) > 0 { 107 119 next = buffer[0] ··· 111 123 rs = nil 112 124 } 113 125 case catchup := <-c.catchup: 114 - job, ok := todo[catchup.user.Uid] 126 + c.maplk.Lock() 127 + job, ok := c.todo[catchup.user.Uid] 128 + // TODO: in the event of receiving a rebase event, we *could* pre-empt all other pending events 115 129 if ok { 116 130 job.catchup = append(job.catchup, catchup) 131 + c.maplk.Unlock() 117 132 break 118 133 } 119 134 120 - job, ok = inProgress[catchup.user.Uid] 135 + job, ok = c.inProgress[catchup.user.Uid] 121 136 if ok { 122 137 job.next = append(job.next, catchup) 138 + c.maplk.Unlock() 123 139 break 124 140 } 125 141 ··· 127 143 act: catchup.user, 128 144 catchup: []*catchupJob{catchup}, 129 145 } 130 - todo[catchup.user.Uid] = cw 146 + c.todo[catchup.user.Uid] = cw 147 + c.maplk.Unlock() 131 148 132 149 if next == nil { 133 150 next = cw ··· 137 154 } 138 155 139 156 case uid := <-c.complete: 140 - job, ok := inProgress[uid] 157 + c.maplk.Lock() 158 + job, ok := c.inProgress[uid] 141 159 if !ok { 142 160 panic("should not be possible to not have a job in progress we receive a completion signal for") 143 161 } 144 - delete(inProgress, uid) 162 + delete(c.inProgress, uid) 145 163 146 164 if len(job.next) > 0 { 147 - todo[uid] = job 165 + c.todo[uid] = job 148 166 job.initScrape = false 149 167 job.catchup = job.next 150 168 job.next = nil ··· 156 174 } 157 175 } 158 176 177 + c.maplk.Unlock() 178 + 159 179 } 160 180 } 161 181 } ··· 206 226 return ctx.Err() 207 227 } 208 228 } 229 + 230 + func (c *CrawlDispatcher) RepoInSlowPath(ctx context.Context, host *models.PDS, uid util.Uid) bool { 231 + c.maplk.Lock() 232 + defer c.maplk.Unlock() 233 + if _, ok := c.todo[uid]; ok { 234 + return true 235 + } 236 + 237 + if _, ok := c.inProgress[uid]; ok { 238 + return true 239 + } 240 + 241 + return false 242 + }
+27
indexer/indexer.go
··· 112 112 113 113 } 114 114 115 + if evt.Rebase { 116 + if err := ix.events.HandleRebase(ctx, evt.User); err != nil { 117 + log.Errorf("failed to handle rebase in events manager: %s", err) 118 + } 119 + } 120 + 115 121 log.Debugw("Sending event", "did", did) 116 122 if err := ix.events.AddEvent(ctx, &events.XRPCStreamEvent{ 117 123 RepoCommit: &comatproto.SyncSubscribeRepos_Commit{ ··· 122 128 Time: time.Now().Format(util.ISO8601), 123 129 Ops: outops, 124 130 TooBig: toobig, 131 + Rebase: evt.Rebase, 125 132 }, 126 133 PrivUid: evt.User, 127 134 }); err != nil { ··· 822 829 curHead, err := ix.repomgr.GetRepoRoot(ctx, ai.Uid) 823 830 if err != nil && !isNotFound(err) { 824 831 return err 832 + } 833 + 834 + var rebase *comatproto.SyncSubscribeRepos_Commit 835 + for _, job := range job.catchup { 836 + if job.evt.Rebase { 837 + rebase = job.evt 838 + break 839 + } 840 + } 841 + if rebase == nil { 842 + for _, job := range job.next { 843 + if job.evt.Rebase { 844 + rebase = job.evt 845 + break 846 + } 847 + } 848 + } 849 + 850 + if rebase != nil { 851 + return ix.repomgr.HandleRebase(ctx, ai.PDS, ai.Uid, ai.Did, (*cid.Cid)(rebase.Prev), (cid.Cid)(rebase.Commit), rebase.Blocks) 825 852 } 826 853 827 854 var host string
+6 -1
pds/handlers.go
··· 792 792 } 793 793 794 794 func (s *Server) handleComAtprotoRepoRebaseRepo(ctx context.Context, body *comatprototypes.RepoRebaseRepo_Input) error { 795 - panic("nyi") 795 + u, err := s.getUser(ctx) 796 + if err != nil { 797 + return err 798 + } 799 + 800 + return s.repoman.DoRebase(ctx, u.ID) 796 801 }
+45
repo/repo.go
··· 160 160 return r.sc.Prev, nil 161 161 } 162 162 163 + func (r *Repo) DataCid() cid.Cid { 164 + return r.sc.Data 165 + } 166 + 163 167 func (r *Repo) SignedCommit() SignedCommit { 164 168 return r.sc 165 169 } ··· 235 239 236 240 r.mst = nmst 237 241 return nil 242 + } 243 + 244 + // truncates history while retaining the same data root 245 + func (r *Repo) Truncate() { 246 + r.sc.Prev = nil 247 + r.repoCid = cid.Undef 238 248 } 239 249 240 250 // creates and writes a new SignedCommit for this repo, with `prev` pointing to old value ··· 381 391 382 392 return mst.DiffTrees(ctx, r.bs, oldTree, curptr) 383 393 } 394 + 395 + func (r *Repo) CopyDataTo(ctx context.Context, bs blockstore.Blockstore) error { 396 + return copyRecCbor(ctx, r.bs, bs, r.sc.Data, make(map[cid.Cid]struct{})) 397 + } 398 + 399 + func copyRecCbor(ctx context.Context, from, to blockstore.Blockstore, c cid.Cid, seen map[cid.Cid]struct{}) error { 400 + if _, ok := seen[c]; ok { 401 + return nil 402 + } 403 + seen[c] = struct{}{} 404 + 405 + blk, err := from.Get(ctx, c) 406 + if err != nil { 407 + return err 408 + } 409 + 410 + if err := to.Put(ctx, blk); err != nil { 411 + return err 412 + } 413 + 414 + var out []cid.Cid 415 + if err := cbg.ScanForLinks(bytes.NewReader(blk.RawData()), func(c cid.Cid) { 416 + out = append(out, c) 417 + }); err != nil { 418 + return err 419 + } 420 + 421 + for _, child := range out { 422 + if err := copyRecCbor(ctx, from, to, child, seen); err != nil { 423 + return err 424 + } 425 + } 426 + 427 + return nil 428 + }
+162 -7
repomgr/repomgr.go
··· 77 77 RepoSlice []byte 78 78 PDS uint 79 79 Ops []RepoOp 80 + Rebase bool 80 81 } 81 82 82 83 type RepoOp struct { ··· 508 509 return ap, nil 509 510 } 510 511 511 - func (rm *RepoManager) HandleExternalUserEvent(ctx context.Context, pdsid uint, uid util.Uid, did string, prev *cid.Cid, carslice []byte) error { 512 - ctx, span := otel.Tracer("repoman").Start(ctx, "HandleExternalUserEvent") 512 + var ErrUncleanRebase = fmt.Errorf("unclean rebase") 513 + 514 + func (rm *RepoManager) HandleRebase(ctx context.Context, pdsid uint, uid util.Uid, did string, prev *cid.Cid, commit cid.Cid, carslice []byte) error { 515 + ctx, span := otel.Tracer("repoman").Start(ctx, "HandleRebase") 513 516 defer span.End() 514 517 515 - log.Infow("HandleExternalUserEvent", "pds", pdsid, "uid", uid, "prev", prev) 518 + log.Infow("HandleRebase", "pds", pdsid, "uid", uid, "commit", commit) 516 519 517 520 unlock := rm.lockUser(ctx, uid) 518 521 defer unlock() 519 522 520 - root, ds, err := rm.cs.ImportSlice(ctx, uid, prev, carslice) 523 + ro, err := rm.cs.ReadOnlySession(uid) 524 + if err != nil { 525 + return err 526 + } 527 + 528 + head, err := rm.cs.GetUserRepoHead(ctx, uid) 529 + if err != nil { 530 + return err 531 + } 532 + 533 + // TODO: do we allow prev to be nil in any case here? 534 + if prev != nil { 535 + if *prev != head { 536 + log.Errorw("rebase 'prev' value did not match our latest head for repo", "did", did, "rprev", prev.String(), "lprev", head.String()) 537 + } 538 + } 539 + 540 + currepo, err := repo.OpenRepo(ctx, ro, head, true) 541 + if err != nil { 542 + return err 543 + } 544 + 545 + olddc := currepo.DataCid() 546 + 547 + root, ds, err := rm.cs.ImportSlice(ctx, uid, nil, carslice) 521 548 if err != nil { 522 549 return fmt.Errorf("importing external carslice: %w", err) 523 550 } ··· 527 554 return fmt.Errorf("opening external user repo (%d, root=%s): %w", uid, root, err) 528 555 } 529 556 530 - repoDid := r.RepoDid() 557 + if r.DataCid() != olddc { 558 + return ErrUncleanRebase 559 + } 531 560 532 - if did != repoDid { 533 - return fmt.Errorf("DID in repo did not match (%q != %q)", did, repoDid) 561 + if err := rm.CheckRepoSig(ctx, r, did); err != nil { 562 + return err 563 + } 564 + 565 + // TODO: this is moderately expensive and currently results in the users 566 + // entire repo being held in memory 567 + if err := r.CopyDataTo(ctx, ds); err != nil { 568 + return err 569 + } 570 + 571 + if err := ds.CloseAsRebase(ctx, root); err != nil { 572 + return fmt.Errorf("finalizing rebase: %w", err) 573 + } 574 + 575 + // TODO: what happens if this update fails? 576 + if err := rm.updateUserRepoHead(ctx, uid, root); err != nil { 577 + return fmt.Errorf("updating user head: %w", err) 578 + } 579 + 580 + if rm.events != nil { 581 + rm.events(ctx, &RepoEvent{ 582 + User: uid, 583 + OldRoot: prev, 584 + NewRoot: root, 585 + Ops: nil, 586 + RepoSlice: carslice, 587 + PDS: pdsid, 588 + Rebase: true, 589 + }) 590 + } 591 + 592 + return nil 593 + } 594 + 595 + func (rm *RepoManager) DoRebase(ctx context.Context, uid util.Uid) error { 596 + ctx, span := otel.Tracer("repoman").Start(ctx, "DoRebase") 597 + defer span.End() 598 + 599 + log.Infow("DoRebase", "uid", uid) 600 + 601 + unlock := rm.lockUser(ctx, uid) 602 + defer unlock() 603 + 604 + ds, err := rm.cs.NewDeltaSession(ctx, uid, nil) 605 + if err != nil { 606 + return err 607 + } 608 + 609 + head, err := rm.cs.GetUserRepoHead(ctx, uid) 610 + if err != nil { 611 + return err 612 + } 613 + 614 + r, err := repo.OpenRepo(ctx, ds, head, true) 615 + if err != nil { 616 + return err 617 + } 618 + 619 + r.Truncate() 620 + 621 + nroot, err := r.Commit(ctx, rm.kmgr.SignForUser) 622 + if err != nil { 623 + return err 624 + } 625 + 626 + if err := ds.CloseAsRebase(ctx, nroot); err != nil { 627 + return fmt.Errorf("finalizing rebase: %w", err) 628 + } 629 + 630 + // outbound car slice should just be the new signed root 631 + buf := new(bytes.Buffer) 632 + if _, err := carstore.WriteCarHeader(buf, nroot); err != nil { 633 + return err 634 + } 635 + robj, err := ds.Get(ctx, nroot) 636 + if err != nil { 637 + return err 638 + } 639 + _, err = carstore.LdWrite(buf, robj.Cid().Bytes(), robj.RawData()) 640 + if err != nil { 641 + return err 642 + } 643 + 644 + if rm.events != nil { 645 + rm.events(ctx, &RepoEvent{ 646 + User: uid, 647 + OldRoot: &head, 648 + NewRoot: nroot, 649 + Ops: nil, 650 + RepoSlice: buf.Bytes(), 651 + PDS: 0, 652 + Rebase: true, 653 + }) 654 + } 655 + 656 + return nil 657 + } 658 + 659 + func (rm *RepoManager) CheckRepoSig(ctx context.Context, r *repo.Repo, expdid string) error { 660 + repoDid := r.RepoDid() 661 + if expdid != repoDid { 662 + return fmt.Errorf("DID in repo did not match (%q != %q)", expdid, repoDid) 534 663 } 535 664 536 665 scom := r.SignedCommit() ··· 542 671 } 543 672 if err := rm.kmgr.VerifyUserSignature(ctx, repoDid, scom.Sig, sb); err != nil { 544 673 return fmt.Errorf("signature check failed: %w", err) 674 + } 675 + 676 + return nil 677 + } 678 + 679 + func (rm *RepoManager) HandleExternalUserEvent(ctx context.Context, pdsid uint, uid util.Uid, did string, prev *cid.Cid, carslice []byte) error { 680 + ctx, span := otel.Tracer("repoman").Start(ctx, "HandleExternalUserEvent") 681 + defer span.End() 682 + 683 + log.Infow("HandleExternalUserEvent", "pds", pdsid, "uid", uid, "prev", prev) 684 + 685 + unlock := rm.lockUser(ctx, uid) 686 + defer unlock() 687 + 688 + root, ds, err := rm.cs.ImportSlice(ctx, uid, prev, carslice) 689 + if err != nil { 690 + return fmt.Errorf("importing external carslice: %w", err) 691 + } 692 + 693 + r, err := repo.OpenRepo(ctx, ds, root, true) 694 + if err != nil { 695 + return fmt.Errorf("opening external user repo (%d, root=%s): %w", uid, root, err) 696 + } 697 + 698 + if err := rm.CheckRepoSig(ctx, r, did); err != nil { 699 + return err 545 700 } 546 701 547 702 var pcid cid.Cid
+76
testing/integ_test.go
··· 1 1 package testing 2 2 3 3 import ( 4 + "bytes" 4 5 "context" 5 6 "fmt" 6 7 "math/rand" ··· 8 9 "time" 9 10 10 11 atproto "github.com/bluesky-social/indigo/api/atproto" 12 + "github.com/bluesky-social/indigo/repo" 13 + "github.com/ipfs/go-cid" 11 14 "github.com/ipfs/go-log/v2" 15 + car "github.com/ipld/go-car" 12 16 "github.com/stretchr/testify/assert" 13 17 ) 14 18 ··· 313 317 last := es2.Next() 314 318 assert.Equal(alice.did, last.RepoCommit.Repo) 315 319 } 320 + 321 + func TestRebase(t *testing.T) { 322 + if testing.Short() { 323 + t.Skip("skipping BGS test in 'short' test mode") 324 + } 325 + assert := assert.New(t) 326 + didr := testPLC(t) 327 + p1 := mustSetupPDS(t, "localhost:9155", ".tpds", didr) 328 + p1.Run(t) 329 + 330 + b1 := mustSetupBGS(t, "localhost:1531", didr) 331 + b1.Run(t) 332 + 333 + p1.RequestScraping(t, b1) 334 + 335 + time.Sleep(time.Millisecond * 50) 336 + 337 + bob := p1.MustNewUser(t, "bob.tpds") 338 + 339 + bob.Post(t, "cats for cats") 340 + bob.Post(t, "i am the king of the world") 341 + bob.Post(t, "the name is bob") 342 + bob.Post(t, "why cant i eat pie") 343 + 344 + time.Sleep(time.Millisecond * 100) 345 + 346 + evts1 := b1.Events(t, 0) 347 + defer evts1.cancel() 348 + 349 + preRebaseEvts := evts1.WaitFor(5) 350 + fmt.Println(preRebaseEvts) 351 + 352 + bob.DoRebase(t) 353 + 354 + rbevt := evts1.Next() 355 + assert.Equal(true, rbevt.RepoCommit.Rebase) 356 + 357 + sc := commitFromSlice(t, rbevt.RepoCommit.Blocks, (cid.Cid)(rbevt.RepoCommit.Commit)) 358 + assert.Nil(sc.Prev) 359 + 360 + lev := preRebaseEvts[4] 361 + oldsc := commitFromSlice(t, lev.RepoCommit.Blocks, (cid.Cid)(lev.RepoCommit.Commit)) 362 + 363 + assert.Equal(sc.Data, oldsc.Data) 364 + 365 + evts2 := b1.Events(t, 0) 366 + afterEvts := evts2.WaitFor(1) 367 + assert.Equal(true, afterEvts[0].RepoCommit.Rebase) 368 + } 369 + 370 + func commitFromSlice(t *testing.T, slice []byte, rcid cid.Cid) *repo.SignedCommit { 371 + carr, err := car.NewCarReader(bytes.NewReader(slice)) 372 + if err != nil { 373 + t.Fatal(err) 374 + } 375 + 376 + for { 377 + blk, err := carr.Next() 378 + if err != nil { 379 + t.Fatal(err) 380 + } 381 + 382 + if blk.Cid() == rcid { 383 + 384 + var sc repo.SignedCommit 385 + if err := sc.UnmarshalCBOR(bytes.NewReader(blk.RawData())); err != nil { 386 + t.Fatal(err) 387 + } 388 + return &sc 389 + } 390 + } 391 + }
+12
testing/utils.go
··· 316 316 } 317 317 } 318 318 319 + func (u *testUser) DoRebase(t *testing.T) { 320 + t.Helper() 321 + 322 + ctx := context.TODO() 323 + err := atproto.RepoRebaseRepo(ctx, u.client, &atproto.RepoRebaseRepo_Input{ 324 + Repo: u.did, 325 + }) 326 + if err != nil { 327 + t.Fatal(err) 328 + } 329 + } 330 + 319 331 func testPLC(t *testing.T) *plc.FakeDid { 320 332 // TODO: just do in memory... 321 333 tdir, err := os.MkdirTemp("", "plcserv")