Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304package spindle
import ( "context" "database/sql" "encoding/json" "errors" "fmt" "slices" "sync" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" "tangled.org/core/api/org_tangled" "tangled.org/core/jetstream" "tangled.org/core/spindle/db" "tangled.org/core/spindle/knotclient" "tangled.org/core/spindle/observability" "tangled.org/core/tapc")
type Ingester func(ctx context.Context, e *jetstream.Event) error
func (s *Spindle) ingest() Ingester { var shards *ingestShards var start sync.Once wipes := make(chan struct{}, accountJobSlots) return func(ctx context.Context, e *jetstream.Event) error { if e.Kind != jetstream.EventKindAccount && e.Kind != jetstream.EventKindCommit { return nil } ticket := jetstream.Hold(ctx) if ticket == nil { err := s.ingestEvent(ctx, e) // ingestEvent already logged it and nothing retries out here if err != nil && e.Kind == jetstream.EventKindCommit && ctx.Err() == nil { return nil } return err } // wipes do io, they skip the shards and hold the ticket until done if e.Kind == jetstream.EventKindAccount { return s.startAccountJob(ctx, e, ticket, wipes) } start.Do(func() { shards = newIngestShards(ctx, func(ctx context.Context, e *jetstream.Event) error { return s.ingestEvent(withMarkOnly(ctx), e) }) }) return shards.dispatch(ctx, e, ticket) }}
const accountJobSlots = 4
// wipes that run past this budget are left to the repo jobsvar accountWipeBudget = 10 * time.Minute
func (s *Spindle) startAccountJob(ctx context.Context, e *jetstream.Event, ticket ingestTicket, slots chan struct{}) error { select { case slots <- struct{}{}: case <-ctx.Done(): ticket.Drop() return ctx.Err() } go func() { defer func() { <-slots }() s.ingestAccountJob(ctx, e, ticket) }() return nil}
func (s *Spindle) ingestAccountJob(ctx context.Context, e *jetstream.Event, ticket ingestTicket) { jobCtx, cancel := context.WithTimeout(ctx, accountWipeBudget) defer cancel() err := s.ingestEvent(jobCtx, e) // a failed wipe already left a wipes row for the repo jobs, and the ticket finishes anyway if err != nil && ctx.Err() == nil { s.l.Error("account event wipe failed, its wipes are retried", "did", e.Did, "err", err) } finishTicket(ctx, ticket, err)}
func (s *Spindle) ingestEvent(ctx context.Context, e *jetstream.Event) error { if e.Kind == jetstream.EventKindAccount { return s.ingestAccount(ctx, e) } if e.Kind != jetstream.EventKindCommit { return nil }
ctx, span := observability.Tracer().Start(ctx, "jetstream.ingest") defer span.End()
if span.IsRecording() { var attrs []attribute.KeyValue if e.Did != "" { attrs = append(attrs, attribute.String(observability.UserDIDKey, e.Did)) } if e.Commit != nil { if e.Commit.Collection != "" { attrs = append(attrs, attribute.String(observability.CollectionKey, e.Commit.Collection)) } if e.Commit.RKey != "" { attrs = append(attrs, attribute.String(observability.RKeyKey, e.Commit.RKey)) } } span.SetAttributes(attrs...) } var err error if evt, ok := jetstreamToTapEvent(e); ok { err = s.processEvent(ctx, evt) }
// a cancelled event stays unconsumed, or the cursor moves past it if err != nil && ctx.Err() != nil { span.SetStatus(codes.Error, "cancelled while processing a jetstream event") return err } if err != nil { s.l.WarnContext(ctx, "failed to process message", "nsid", e.Commit.Collection, "did", e.Did, "rkey", e.Commit.RKey, "err", err) s.metrics.RecordEventIngestion("jetstream", "error") span.SetStatus(codes.Error, "failed to process jetstream event") return err } s.metrics.RecordEventIngestion("jetstream", "success") span.SetStatus(codes.Ok, "success") return nil}
// the firehose sees every repo, so each handler drops what spindle doesn't trackfunc (s *Spindle) processEvent(ctx context.Context, evt tapc.Event) error { if evt.Type != tapc.EvtRecord || evt.Record == nil { return nil } rec := evt.Record var handle func(context.Context, *tapc.RecordEventData) error switch rec.Collection.String() { case org_tangled.RepoManifestNSID: handle = s.onManifest case org_tangled.GraphCollaboratorNSID: handle = s.onCollaborator case org_tangled.RepoDeclarationNSID: handle = s.onDeclaration case org_tangled.TrackTicketNSID: handle = s.onTicket case org_tangled.ReviewPatchNSID: handle = s.onPatch default: return nil } source := "backfill" if rec.Live { source = "live" } if err := handle(ctx, rec); err != nil { s.metrics.RecordRepoRecord(source, "failed") return err } s.metrics.RecordRepoRecord(source, "applied") return nil}
func (s *Spindle) onManifest(ctx context.Context, rec *tapc.RecordEventData) error { if rec.Rkey != knotclient.ManifestRkey { return nil } known, err := s.knowsRepo(rec.Did) if err != nil { return err } if !known { if rec.Action == tapc.RecordDeleteAction { return nil } var manifest org_tangled.RepoManifest if err := json.Unmarshal(rec.Record, &manifest); err != nil { s.l.Warn("skipping invalid manifest record", "repo", rec.Did, "err", err) return nil } if manifest.Spindle == nil || *manifest.Spindle != s.cfg.Server.Did().String() { return nil } // only enrolled owners get knot and pds reads, since anyone can point a manifest at this spindle if manifest.Declaration == nil { return nil } owner, err := syntax.ParseDID(manifest.Declaration.Owner) if err != nil { return nil } if enrolled, err := db.Enrolls(s.db, &s.cfg.Server, owner); err != nil || !enrolled { return err } } return s.refreshForEvent(ctx, rec, rec.Did)}
func (s *Spindle) onCollaborator(ctx context.Context, rec *tapc.RecordEventData) error { known, err := s.knowsRepo(rec.Did) if err != nil || !known { return err } if markOnly(ctx) { if err := s.markRecord(rec); err != nil { return err } } return s.refreshForEvent(ctx, rec, rec.Did)}
// delete events carry no record, so the repo has to be found by owner and rkeyfunc (s *Spindle) onDeclaration(ctx context.Context, rec *tapc.RecordEventData) error { if rec.Action == tapc.RecordDeleteAction { repo, err := s.db.GetRepoByOwnerRkey(rec.Did, rec.Rkey) if errors.Is(err, sql.ErrNoRows) { return nil } if err != nil { return err } // old rows can have an empty repo did, so there's nothing to refresh if repo.RepoDid == "" { return nil } return s.refreshForEvent(ctx, rec, repo.RepoDid) } var declaration org_tangled.RepoDeclaration if err := json.Unmarshal(rec.Record, &declaration); err != nil { s.l.Warn("skipping invalid declaration record", "owner", rec.Did, "err", err) return nil } repoDid, err := syntax.ParseDID(declaration.Repo) if err != nil { s.l.Warn("skipping declaration with a malformed repo did", "owner", rec.Did, "repo", declaration.Repo) return nil } rows, err := s.db.ReposByDid(repoDid) if err != nil { return err } if len(rows) > 0 { // anyone can write a declaration naming a served repo, and every refresh costs round trips if !slices.ContainsFunc(rows, func(r db.Repo) bool { return r.Owner == rec.Did }) { return nil } } else if enrolled, err := db.Enrolls(s.db, &s.cfg.Server, rec.Did); err != nil || !enrolled { return err } return s.refreshForEvent(ctx, rec, repoDid)}
func jetstreamToTapEvent(e *jetstream.Event) (tapc.Event, bool) { if e.Commit == nil { return tapc.Event{}, false } did, err := syntax.ParseDID(e.Did) if err != nil { return tapc.Event{}, false } var action tapc.RecordAction switch e.Commit.Operation { case jetstream.CommitOperationCreate: action = tapc.RecordCreateAction case jetstream.CommitOperationUpdate: action = tapc.RecordUpdateAction case jetstream.CommitOperationDelete: action = tapc.RecordDeleteAction default: return tapc.Event{}, false } return tapc.Event{ Type: tapc.EvtRecord, Record: &tapc.RecordEventData{ Did: did, Rkey: syntax.RecordKey(e.Commit.RKey), Collection: syntax.NSID(e.Commit.Collection), Rev: e.Commit.Rev, Action: action, Record: e.Commit.Record, // jetstream is only used for live Live: true, }, }, true}
func (s *Spindle) ingestAccount(ctx context.Context, e *jetstream.Event) error { if e.Account == nil || e.Account.Active { return nil } status := "unknown" if e.Account.Status != nil { status = *e.Account.Status } did, err := syntax.ParseDID(e.Account.Did) if err != nil { return fmt.Errorf("parsing account did: %w", err) } s.l.Warn("account went inactive, purging its repos", "did", did, "status", status) return s.WipeOwner(ctx, did, "account "+status)}