From b3ee1d8d5e894f6413b510f1f0fb351ead270220 Mon Sep 17 00:00:00 2001 From: Lewis Date: Mon, 15 Jun 2026 10:17:48 +0300 Subject: [PATCH] appview/state: ingest knot ACL events into roster Lewis: May this revision serve well! --- appview/db/vouch.go | 2 +- appview/state/knotstream.go | 125 +++++++++++- appview/state/knotstream_test.go | 333 +++++++++++++++++++++++++++++++ appview/state/state.go | 2 +- 4 files changed, 457 insertions(+), 5 deletions(-) create mode 100644 appview/state/knotstream_test.go diff --git a/appview/db/vouch.go b/appview/db/vouch.go index 695fc471..ed0c18bf 100644 --- a/appview/db/vouch.go +++ b/appview/db/vouch.go @@ -343,7 +343,7 @@ func SkipVouchSuggestion(e Execer, did, subjectDid string) error { } // priority: -// 1. collaborator invites sent +// 1. collaborator invites sent - NOTE with knot-owned events not mentioning *who* is doing the adding of collab, we can't know who to suggest a vouch to. // 2. knot member invites sent // 3. PR authors on FOO's repositories // 4. issue authors on FOO's repositories diff --git a/appview/state/knotstream.go b/appview/state/knotstream.go index 17352dec..5cf824ba 100644 --- a/appview/state/knotstream.go +++ b/appview/state/knotstream.go @@ -16,6 +16,7 @@ import ( "tangled.org/core/api/tangled" "tangled.org/core/appview/config" "tangled.org/core/appview/db" + "tangled.org/core/appview/knotacl" "tangled.org/core/appview/knotcompat" "tangled.org/core/appview/models" "tangled.org/core/appview/sites" @@ -33,7 +34,16 @@ import ( "github.com/posthog/posthog-go" ) -func Knotstream(ctx context.Context, c *config.Config, d *db.DB, enforcer *rbac.Enforcer, posthog posthog.Client, notifier notify.Notifier, cfClient *cloudflare.Client) (*ec.Consumer, error) { +type aclRoster interface { + AddKnotMember(host string, subject syntax.DID, cursor knotacl.Cursor) error + RemoveKnotMember(host string, subject syntax.DID, cursor knotacl.Cursor) error + AddCollaborator(repoDid, subject syntax.DID, cursor knotacl.Cursor) error + RemoveCollaborator(repoDid, subject syntax.DID, cursor knotacl.Cursor) error + InvalidateMembers(host string) + InvalidateCollaborators(host, repoDid string) +} + +func Knotstream(ctx context.Context, c *config.Config, d *db.DB, acl *knotacl.Service, enforcer *rbac.Enforcer, posthog posthog.Client, notifier notify.Notifier, cfClient *cloudflare.Client) (*ec.Consumer, error) { knots, err := db.GetRegistrations(d, orm.FilterIsNot("registered", "null")) if err != nil { return nil, err @@ -47,7 +57,7 @@ func Knotstream(ctx context.Context, c *config.Config, d *db.DB, enforcer *rbac. return bootstrapStream( ctx, "knotstream", ec.KindKnot, hosts, c.Redis.Addr, c.Knotstream, c.Core.Dev, - knotIngester(d, enforcer, posthog, notifier, c.Core.Dev, c, cfClient), + knotIngester(d, acl, enforcer, posthog, notifier, c.Core.Dev, c, cfClient), ), nil } @@ -65,7 +75,7 @@ func resolveRepo(d *db.DB, repoDid *string, ownerDid, repoName string) (*models. return &repos[0], nil } -func knotIngester(d *db.DB, enforcer *rbac.Enforcer, posthog posthog.Client, notifier notify.Notifier, dev bool, c *config.Config, cfClient *cloudflare.Client) ec.ProcessFunc { +func knotIngester(d *db.DB, acl aclRoster, enforcer *rbac.Enforcer, posthog posthog.Client, notifier notify.Notifier, dev bool, c *config.Config, cfClient *cloudflare.Client) ec.ProcessFunc { return func(ctx context.Context, source ec.Source, msg eventstream.Event) error { switch msg.Nsid { case tangled.GitRefUpdateNSID: @@ -74,12 +84,121 @@ func knotIngester(d *db.DB, enforcer *rbac.Enforcer, posthog posthog.Client, not return ingestPipeline(d, source, msg) case knotdb.RepoDIDAssignNSID: return ingestDIDAssign(d, enforcer, source, msg, ctx) + case knotdb.KnotMemberUpdateNSID: + return ingestKnotMemberUpdate(acl, source, msg) + case knotdb.RepoCollaboratorUpdateNSID: + return ingestCollaboratorUpdate(ctx, d, acl, source, msg) } return nil } } +const ( + aclIngestAttempts = 3 + aclIngestBackoff = 50 * time.Millisecond +) + +func withAclRetry(attempts int, backoff time.Duration, op func() error) error { + if err := op(); err == nil || attempts <= 1 { + return err + } + time.Sleep(backoff) + return withAclRetry(attempts-1, backoff, op) +} + +func ingestKnotMemberUpdate(acl aclRoster, source ec.Source, msg eventstream.Event) error { + var rec knotdb.KnotMemberUpdate + if err := json.Unmarshal(msg.EventJson, &rec); err != nil { + return fmt.Errorf("unmarshal memberUpdate: %w", err) + } + + subject, err := syntax.ParseDID(rec.Subject) + if err != nil { + return fmt.Errorf("memberUpdate bad subject %q: %w", rec.Subject, err) + } + + cursor := knotacl.Cursor(msg.Created) + switch rec.Op { + case knotdb.AclOpAdd: + err = withAclRetry(aclIngestAttempts, aclIngestBackoff, func() error { + return acl.AddKnotMember(source.Host, subject, cursor) + }) + case knotdb.AclOpRemove: + err = withAclRetry(aclIngestAttempts, aclIngestBackoff, func() error { + return acl.RemoveKnotMember(source.Host, subject, cursor) + }) + default: + return fmt.Errorf("memberUpdate unknown op %q", rec.Op) + } + + if err != nil { + acl.InvalidateMembers(source.Host) + } + return err +} + +func ingestCollaboratorUpdate(ctx context.Context, d *db.DB, acl aclRoster, source ec.Source, msg eventstream.Event) error { + var rec knotdb.RepoCollaboratorUpdate + if err := json.Unmarshal(msg.EventJson, &rec); err != nil { + return fmt.Errorf("unmarshal collaboratorUpdate: %w", err) + } + + subject, err := syntax.ParseDID(rec.Subject) + if err != nil { + return fmt.Errorf("collaboratorUpdate bad subject %q: %w", rec.Subject, err) + } + repoDid, err := syntax.ParseDID(rec.Repo) + if err != nil { + return fmt.Errorf("collaboratorUpdate bad repo %q: %w", rec.Repo, err) + } + + cursor := knotacl.Cursor(msg.Created) + switch rec.Op { + case knotdb.AclOpAdd: + err = withAclRetry(aclIngestAttempts, aclIngestBackoff, func() error { + owned, err := repoOwnedBySource(ctx, d, source, repoDid, subject) + if err != nil || !owned { + return err + } + return acl.AddCollaborator(repoDid, subject, cursor) + }) + case knotdb.AclOpRemove: + err = withAclRetry(aclIngestAttempts, aclIngestBackoff, func() error { + owned, err := repoOwnedBySource(ctx, d, source, repoDid, subject) + if err != nil || !owned { + return err + } + return acl.RemoveCollaborator(repoDid, subject, cursor) + }) + default: + return fmt.Errorf("collaboratorUpdate unknown op %q", rec.Op) + } + + if err != nil { + acl.InvalidateCollaborators(source.Host, repoDid.String()) + } + return err +} + +func repoOwnedBySource(ctx context.Context, d *db.DB, source ec.Source, repoDid, subject syntax.DID) (bool, error) { + repo, err := db.GetRepoByDid(d, repoDid.String()) + if errors.Is(err, sql.ErrNoRows) { + log.FromContext(ctx).Warn("collaboratorUpdate for unindexed repo, skipping until reconcile", + "repo_did", repoDid, "subject", subject) + return false, nil + } + if err != nil { + return false, err + } + if repo.Knot != source.Host { + log.FromContext(ctx).Warn("collaboratorUpdate for a repo this knot does not host, dropping", + "repo_did", repoDid, "subject", subject, "claimed_by", source.Host, "owner", repo.Knot) + return false, nil + } + return true, nil +} + // TODO(boltless): remove this. knotmirror should do all sort of indexing func ingestRefUpdate(ctx context.Context, d *db.DB, enforcer *rbac.Enforcer, pc posthog.Client, notifier notify.Notifier, dev bool, c *config.Config, cfClient *cloudflare.Client, source ec.Source, msg eventstream.Event) error { logger := log.FromContext(ctx) diff --git a/appview/state/knotstream_test.go b/appview/state/knotstream_test.go new file mode 100644 index 00000000..f00938dd --- /dev/null +++ b/appview/state/knotstream_test.go @@ -0,0 +1,333 @@ +package state + +import ( + "context" + "encoding/json" + "errors" + "path/filepath" + "testing" + + "github.com/bluesky-social/indigo/atproto/syntax" + + "tangled.org/core/appview/db" + "tangled.org/core/appview/knotacl" + "tangled.org/core/appview/models" + ec "tangled.org/core/eventconsumer" + "tangled.org/core/eventstream" + knotdb "tangled.org/core/knotserver/db" +) + +const ( + aclTestHost = "knot.nel.pet" + aclTestRepoDid = "did:plc:limpet" + aclTestOwner = "did:plc:akshay" + aclTestSubject = "did:plc:boltless" +) + +type memberCall struct { + host string + subject string +} + +type collabCall struct { + repoDid string + subject string +} + +type recordingAcl struct { + memberAdd []memberCall + memberRemove []memberCall + collabAdd []collabCall + collabRemove []collabCall + membersInvalid []string + collabsInvalid []collabCall +} + +func (r *recordingAcl) AddKnotMember(host string, subject syntax.DID, cursor knotacl.Cursor) error { + r.memberAdd = append(r.memberAdd, memberCall{host, subject.String()}) + return nil +} + +func (r *recordingAcl) RemoveKnotMember(host string, subject syntax.DID, cursor knotacl.Cursor) error { + r.memberRemove = append(r.memberRemove, memberCall{host, subject.String()}) + return nil +} + +func (r *recordingAcl) AddCollaborator(repoDid, subject syntax.DID, cursor knotacl.Cursor) error { + r.collabAdd = append(r.collabAdd, collabCall{repoDid.String(), subject.String()}) + return nil +} + +func (r *recordingAcl) RemoveCollaborator(repoDid, subject syntax.DID, cursor knotacl.Cursor) error { + r.collabRemove = append(r.collabRemove, collabCall{repoDid.String(), subject.String()}) + return nil +} + +func (r *recordingAcl) InvalidateMembers(host string) { + r.membersInvalid = append(r.membersInvalid, host) +} + +func (r *recordingAcl) InvalidateCollaborators(host, repoDid string) { + r.collabsInvalid = append(r.collabsInvalid, collabCall{repoDid, host}) +} + +type flakyAcl struct { + failsLeft int + calls int + membersInvalid int + collabsInvalid int +} + +func (a *flakyAcl) try() error { + a.calls++ + if a.failsLeft > 0 { + a.failsLeft-- + return errors.New("transient store error") + } + return nil +} + +func (a *flakyAcl) AddKnotMember(host string, subject syntax.DID, cursor knotacl.Cursor) error { + return a.try() +} +func (a *flakyAcl) RemoveKnotMember(host string, subject syntax.DID, cursor knotacl.Cursor) error { + return a.try() +} +func (a *flakyAcl) AddCollaborator(repoDid, subject syntax.DID, cursor knotacl.Cursor) error { + return a.try() +} +func (a *flakyAcl) RemoveCollaborator(repoDid, subject syntax.DID, cursor knotacl.Cursor) error { + return a.try() +} +func (a *flakyAcl) InvalidateMembers(host string) { a.membersInvalid++ } +func (a *flakyAcl) InvalidateCollaborators(host, repo string) { a.collabsInvalid++ } + +func aclTestDB(t *testing.T) *db.DB { + t.Helper() + d, err := db.Make(context.Background(), filepath.Join(t.TempDir(), "appview.db")) + if err != nil { + t.Fatalf("db.Make: %v", err) + } + t.Cleanup(func() { d.Close() }) + return d +} + +func seedAclRepo(t *testing.T, d *db.DB) { + t.Helper() + tx, err := d.Begin() + if err != nil { + t.Fatalf("begin: %v", err) + } + if err := db.AddRepo(tx, &models.Repo{ + Did: aclTestOwner, + Knot: aclTestHost, + RepoDid: aclTestRepoDid, + Name: "anemone", + }); err != nil { + t.Fatalf("AddRepo: %v", err) + } + if err := tx.Commit(); err != nil { + t.Fatalf("commit: %v", err) + } +} + +func memberEvent(t *testing.T, op knotdb.AclOp, subject string) eventstream.Event { + t.Helper() + payload, err := json.Marshal(knotdb.KnotMemberUpdate{Op: op, Subject: subject}) + if err != nil { + t.Fatalf("marshal memberUpdate: %v", err) + } + return eventstream.Event{Rkey: "evt", Nsid: knotdb.KnotMemberUpdateNSID, EventJson: payload} +} + +func collabEvent(t *testing.T, op knotdb.AclOp, subject, repoDid string) eventstream.Event { + t.Helper() + payload, err := json.Marshal(knotdb.RepoCollaboratorUpdate{Op: op, Subject: subject, Repo: repoDid}) + if err != nil { + t.Fatalf("marshal collaboratorUpdate: %v", err) + } + return eventstream.Event{Rkey: "evt", Nsid: knotdb.RepoCollaboratorUpdateNSID, EventJson: payload} +} + +func TestIngestKnotMemberUpdate_DispatchesAddThenRemove(t *testing.T) { + acl := &recordingAcl{} + source := ec.Source{Kind: ec.KindKnot, Host: aclTestHost} + + if err := ingestKnotMemberUpdate(acl, source, memberEvent(t, knotdb.AclOpAdd, aclTestSubject)); err != nil { + t.Fatalf("add: %v", err) + } + if err := ingestKnotMemberUpdate(acl, source, memberEvent(t, knotdb.AclOpRemove, aclTestSubject)); err != nil { + t.Fatalf("remove: %v", err) + } + + if len(acl.memberAdd) != 1 || acl.memberAdd[0] != (memberCall{aclTestHost, aclTestSubject}) { + t.Errorf("memberAdd = %v, want one add scoped to the source host", acl.memberAdd) + } + if len(acl.memberRemove) != 1 || acl.memberRemove[0] != (memberCall{aclTestHost, aclTestSubject}) { + t.Errorf("memberRemove = %v, want one remove scoped to the source host", acl.memberRemove) + } +} + +func TestIngestKnotMemberUpdate_UnknownOpErrors(t *testing.T) { + acl := &recordingAcl{} + source := ec.Source{Kind: ec.KindKnot, Host: aclTestHost} + if err := ingestKnotMemberUpdate(acl, source, memberEvent(t, knotdb.AclOp("bogus"), aclTestSubject)); err == nil { + t.Fatal("an unknown op must be rejected") + } + if len(acl.memberAdd)+len(acl.memberRemove) != 0 { + t.Errorf("an unknown op must not reach the roster: %+v", acl) + } +} + +func TestIngestKnotMemberUpdate_BadSubjectErrors(t *testing.T) { + acl := &recordingAcl{} + source := ec.Source{Kind: ec.KindKnot, Host: aclTestHost} + if err := ingestKnotMemberUpdate(acl, source, memberEvent(t, knotdb.AclOpAdd, "not-a-did")); err == nil { + t.Fatal("a malformed subject DID must be rejected") + } + if len(acl.memberAdd) != 0 { + t.Errorf("a malformed subject must not reach the roster: %v", acl.memberAdd) + } +} + +func TestIngestCollaboratorUpdate_DispatchesAddThenRemove(t *testing.T) { + ctx := context.Background() + d := aclTestDB(t) + seedAclRepo(t, d) + acl := &recordingAcl{} + source := ec.Source{Kind: ec.KindKnot, Host: aclTestHost} + + if err := ingestCollaboratorUpdate(ctx, d, acl, source, collabEvent(t, knotdb.AclOpAdd, aclTestSubject, aclTestRepoDid)); err != nil { + t.Fatalf("add: %v", err) + } + if err := ingestCollaboratorUpdate(ctx, d, acl, source, collabEvent(t, knotdb.AclOpRemove, aclTestSubject, aclTestRepoDid)); err != nil { + t.Fatalf("remove: %v", err) + } + + if len(acl.collabAdd) != 1 || acl.collabAdd[0] != (collabCall{aclTestRepoDid, aclTestSubject}) { + t.Errorf("collabAdd = %v, want one add for the repo", acl.collabAdd) + } + if len(acl.collabRemove) != 1 || acl.collabRemove[0] != (collabCall{aclTestRepoDid, aclTestSubject}) { + t.Errorf("collabRemove = %v, want one remove for the repo", acl.collabRemove) + } +} + +func TestIngestCollaboratorUpdate_UnindexedRepoSkips(t *testing.T) { + ctx := context.Background() + d := aclTestDB(t) + acl := &recordingAcl{} + source := ec.Source{Kind: ec.KindKnot, Host: aclTestHost} + + if err := ingestCollaboratorUpdate(ctx, d, acl, source, collabEvent(t, knotdb.AclOpAdd, aclTestSubject, aclTestRepoDid)); err != nil { + t.Fatalf("add for unindexed repo must not error, got: %v", err) + } + if len(acl.collabAdd) != 0 { + t.Errorf("an add for an unindexed repo must not reach the roster: %v", acl.collabAdd) + } +} + +func TestIngestCollaboratorUpdate_ForeignKnotDropped(t *testing.T) { + ctx := context.Background() + d := aclTestDB(t) + seedAclRepo(t, d) + acl := &recordingAcl{} + source := ec.Source{Kind: ec.KindKnot, Host: "barnacle.nel.pet"} + + if err := ingestCollaboratorUpdate(ctx, d, acl, source, collabEvent(t, knotdb.AclOpAdd, aclTestSubject, aclTestRepoDid)); err != nil { + t.Fatalf("a foreign-knot collaboratorUpdate must be dropped, not error: %v", err) + } + if len(acl.collabAdd) != 0 { + t.Errorf("a knot that does not host the repo must not mutate its collaborators: %v", acl.collabAdd) + } + + if err := ingestCollaboratorUpdate(ctx, d, acl, source, collabEvent(t, knotdb.AclOpRemove, aclTestSubject, aclTestRepoDid)); err != nil { + t.Fatalf("a foreign-knot remove must be dropped, not error: %v", err) + } + if len(acl.collabRemove) != 0 { + t.Errorf("a knot that does not host the repo must not remove its collaborators: %v", acl.collabRemove) + } +} + +func TestIngestCollaboratorUpdate_BadDidErrors(t *testing.T) { + ctx := context.Background() + d := aclTestDB(t) + acl := &recordingAcl{} + source := ec.Source{Kind: ec.KindKnot, Host: aclTestHost} + if err := ingestCollaboratorUpdate(ctx, d, acl, source, collabEvent(t, knotdb.AclOpAdd, "not-a-did", aclTestRepoDid)); err == nil { + t.Fatal("a malformed subject DID must be rejected") + } + if len(acl.collabAdd) != 0 { + t.Errorf("a malformed subject must not reach the roster: %v", acl.collabAdd) + } +} + +func TestIngestKnotMemberUpdate_RetriesTransientThenSucceeds(t *testing.T) { + acl := &flakyAcl{failsLeft: aclIngestAttempts - 1} + source := ec.Source{Kind: ec.KindKnot, Host: aclTestHost} + + if err := ingestKnotMemberUpdate(acl, source, memberEvent(t, knotdb.AclOpAdd, aclTestSubject)); err != nil { + t.Fatalf("a transient store error within the retry budget must recover, got: %v", err) + } + if acl.calls != aclIngestAttempts { + t.Errorf("calls = %d, want %d; the write must retry until it lands", acl.calls, aclIngestAttempts) + } +} + +func TestIngestKnotMemberUpdate_GivesUpAfterAttempts(t *testing.T) { + acl := &flakyAcl{failsLeft: aclIngestAttempts + 5} + source := ec.Source{Kind: ec.KindKnot, Host: aclTestHost} + + if err := ingestKnotMemberUpdate(acl, source, memberEvent(t, knotdb.AclOpAdd, aclTestSubject)); err == nil { + t.Fatal("a persistent store error must surface so the failure is logged") + } + if acl.calls != aclIngestAttempts { + t.Errorf("calls = %d, want %d; the retry must be bounded", acl.calls, aclIngestAttempts) + } + if acl.membersInvalid != 1 { + t.Errorf("membersInvalid = %d, want 1; a dropped delta must invalidate the scope so the next read reconciles instead of waiting out the TTL", acl.membersInvalid) + } +} + +func TestIngestCollaboratorUpdate_InvalidatesScopeOnGiveUp(t *testing.T) { + ctx := context.Background() + d := aclTestDB(t) + seedAclRepo(t, d) + acl := &flakyAcl{failsLeft: aclIngestAttempts + 5} + source := ec.Source{Kind: ec.KindKnot, Host: aclTestHost} + + if err := ingestCollaboratorUpdate(ctx, d, acl, source, collabEvent(t, knotdb.AclOpAdd, aclTestSubject, aclTestRepoDid)); err == nil { + t.Fatal("a persistent store error must surface so the failure is logged") + } + if acl.collabsInvalid != 1 { + t.Errorf("collabsInvalid = %d, want 1; a dropped delta must invalidate the scope", acl.collabsInvalid) + } +} + +func TestIngestKnotMemberUpdate_NoInvalidateOnSuccess(t *testing.T) { + acl := &flakyAcl{failsLeft: aclIngestAttempts - 1} + source := ec.Source{Kind: ec.KindKnot, Host: aclTestHost} + + if err := ingestKnotMemberUpdate(acl, source, memberEvent(t, knotdb.AclOpAdd, aclTestSubject)); err != nil { + t.Fatalf("a recoverable delta must not error: %v", err) + } + if acl.membersInvalid != 0 { + t.Errorf("membersInvalid = %d, want 0; a delta that lands must not force a reconcile", acl.membersInvalid) + } +} + +func TestIngestCollaboratorUpdate_StoreErrorPropagates(t *testing.T) { + ctx := context.Background() + d := aclTestDB(t) + if err := d.Close(); err != nil { + t.Fatalf("close: %v", err) + } + acl := &recordingAcl{} + source := ec.Source{Kind: ec.KindKnot, Host: aclTestHost} + + if err := ingestCollaboratorUpdate(ctx, d, acl, source, collabEvent(t, knotdb.AclOpAdd, aclTestSubject, aclTestRepoDid)); err == nil { + t.Fatal("a store error on the repo lookup must surface, not be swallowed as an unindexed-repo skip") + } + if len(acl.collabAdd) != 0 { + t.Errorf("a failed repo lookup must not reach the roster: %v", acl.collabAdd) + } +} diff --git a/appview/state/state.go b/appview/state/state.go index 529abb33..e73ff28d 100644 --- a/appview/state/state.go +++ b/appview/state/state.go @@ -215,7 +215,7 @@ func Make(ctx context.Context, config *config.Config) (*State, error) { } } - knotstream, err := Knotstream(ctx, config, d, enforcer, posthog, notifier, cfClient) + knotstream, err := Knotstream(ctx, config, d, aclService, enforcer, posthog, notifier, cfClient) if err != nil { return nil, fmt.Errorf("failed to start knotstream consumer: %w", err) } -- 2.51.2