From ece4511ac7ca04f5d02052f1ce1df54066c37fb0 Mon Sep 17 00:00:00 2001 From: Bretton Date: Thu, 13 Aug 2026 06:30:01 -0700 Subject: [PATCH] =?UTF-8?q?wip(task15):=20cycles=20F-I=20=E2=80=94=20deliv?= =?UTF-8?q?ery=20workers,=20retry=20taxonomy,=20causal=20gating,=20inbox?= =?UTF-8?q?=20discovery,=20kill=20switches?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Worker owns retry (SendActivityAs is single-shot returning HTTPError{StatusCode,Body}); duplicate-activity→delivered, consent recheck exempts Delete/Undo (retraction asymmetry), vote callbacks clear/flip outbound_votes. Causal gate is a post-claim check (bridge-origin parents gate on accepted_at; fediverse always eligible); inbox re-resolves once on 401/404/410; kill switches park (never poison). Outer acceptance test green. Co-Authored-By: Claude Fable 5 --- internal/ap/client.go | 46 +- internal/outbound/causal_gating_test.go | 159 +++++++ internal/outbound/inbox_discovery_test.go | 132 ++++++ internal/outbound/inbox_resolver.go | 81 ++++ internal/outbound/outbound.go | 47 +- internal/outbound/worker.go | 531 ++++++++++++++++++++-- internal/outbound/worker_test.go | 518 +++++++++++++++++++++ internal/personas/personas.go | 7 + internal/store/interfaces.go | 15 + internal/store/models.go | 8 + internal/store/outbound_deliveries.go | 14 + internal/store/outbound_objects.go | 9 +- internal/store/outbound_votes.go | 16 + 13 files changed, 1549 insertions(+), 34 deletions(-) create mode 100644 internal/outbound/causal_gating_test.go create mode 100644 internal/outbound/inbox_discovery_test.go create mode 100644 internal/outbound/inbox_resolver.go create mode 100644 internal/outbound/worker_test.go diff --git a/internal/ap/client.go b/internal/ap/client.go index 6f7e769..ea4132e 100644 --- a/internal/ap/client.go +++ b/internal/ap/client.go @@ -91,6 +91,12 @@ func (e *CollectionTruncatedError) Unwrap() error { return ErrCollectionTruncate type HTTPError struct { URL string StatusCode int + // Body is a bounded excerpt of the response body, populated on delivery + // POSTs so the caller can classify peer-specific signals (task 15's worker + // keys Lemmy's duplicate-activity response — a 400 whose body names an + // already-received activity — as DELIVERED, not poisoned). Empty when the + // body was not captured. + Body string } func (e HTTPError) Error() string { @@ -600,13 +606,47 @@ func (c *Client) SendActivity(ctx context.Context, inboxURL string, activity any // SendActivityAs signed-POSTs an activity to a remote inbox using the GIVEN // per-actor signer, not the client's configured Signer (which stays reserved // for the service actor's Follow/Undo). Task 15's delivery worker signs each -// activity as the persona that authored the record. Retry/status handling -// mirrors SendActivity. +// activity as the persona that authored the record. +// +// Unlike SendActivity, this is a SINGLE POST with no internal retry loop: the +// task-15 Worker owns retry, backoff and the poison/attempt-cap policy, and it +// needs to see each response to classify it (Lemmy's duplicate-activity 400 is +// a SUCCESS, a 404 triggers inbox re-resolution). A non-2xx response is +// returned as an HTTPError carrying the status AND a bounded body excerpt, so +// the worker can read Lemmy's "already received" body; a transport failure is +// any other error. func (c *Client) SendActivityAs(ctx context.Context, signer *Signer, inboxURL string, activity any) error { if signer == nil { return errors.NewValidationError("signer", "SendActivityAs requires a per-actor Signer") } - return c.sendActivityWith(ctx, signer, inboxURL, activity) + payload, err := json.Marshal(activity) + if err != nil { + return fmt.Errorf("ap: encode activity: %w", err) + } + if err := c.waitForHost(ctx, inboxURL); err != nil { + return err + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, inboxURL, bytes.NewReader(payload)) + if err != nil { + return fmt.Errorf("ap: build POST %s: %w", inboxURL, err) + } + req.Header.Set("User-Agent", c.userAgent) + req.Header.Set("Content-Type", ContentTypeActivityJSON) + if err := signer.SignRequest(req, payload); err != nil { + return err + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("ap: POST %s: %w", inboxURL, err) + } + defer func() { _ = resp.Body.Close() }() + body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + return nil + } + return HTTPError{URL: inboxURL, StatusCode: resp.StatusCode, Body: string(body)} } // sendActivityWith is the shared POST loop: it marshals the activity once, diff --git a/internal/outbound/causal_gating_test.go b/internal/outbound/causal_gating_test.go new file mode 100644 index 0000000..8eb2401 --- /dev/null +++ b/internal/outbound/causal_gating_test.go @@ -0,0 +1,159 @@ +package outbound + +import ( + "context" + "database/sql" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/store" +) + +// Task 15 cycle G: causal gating. A reply must never be delivered before the +// thing it replies to — but ONLY when that parent is a BRIDGE-origin object we +// are still delivering. A reply to a Lemmy post (fediverse-origin: already on +// Lemmy) must ALWAYS be eligible. Getting this wrong poisons every reply to a +// Lemmy post, so the bridge-vs-fediverse distinction is the crux. + +const gParentATURI = "at://" + wCommunityDID + "/social.coves.community.postv2/3lzparentpost" + +// seedBridgeParent writes an outbound_objects row for the parent (a bridge-origin +// object Tidepool federates). accepted=false leaves accepted_at NULL. +func seedBridgeParent(t *testing.T, conn *sql.DB, accepted bool) { + t.Helper() + ctx := context.Background() + _, err := store.NewOutboundObjects(conn).Upsert(ctx, store.OutboundObject{ + ATURI: gParentATURI, + APObjectID: "https://coves.social/ap/object/" + wCommunityDID + "/social.coves.community.postv2/3lzparentpost", + CommunityDID: wCommunityDID, + CommunityAPID: wCommunityAPID, + TranslatedSnapshot: []byte(`{"type":"Page"}`), + }) + require.NoError(t, err) + if accepted { + require.NoError(t, store.NewOutboundObjects(conn).SetAccepted(ctx, gParentATURI)) + } +} + +func TestCausalGating_BridgeParentUnacceptedIsIneligible(t *testing.T) { + conn := workerTestDB(t) + seedWorkerActor(t, conn, true, false) + seedBridgeParent(t, conn, false) // accepted_at NULL + id := seedDelivery(t, conn, "Create", gParentATURI, createPayload("x")) + + sender := &fakeSender{} + w := newWorker(t, conn, sender, nil) + _, err := w.DeliverNext(context.Background()) + require.NoError(t, err) + + assert.Zero(t, sender.count(), + "a reply to a not-yet-accepted BRIDGE parent must NOT be delivered") + assert.Equal(t, store.DeliveryStatePending, getDelivery(t, conn, id).State, + "it stays pending (ineligible), waiting for the parent to land") + + // Flip the parent to accepted: the SAME delivery must now go out. This is + // what keeps the negative non-vacuous — the gate opens, it delivers. + require.NoError(t, store.NewOutboundObjects(conn).SetAccepted(context.Background(), gParentATURI)) + worked, err := w.DeliverNext(context.Background()) + require.NoError(t, err) + assert.True(t, worked) + assert.Equal(t, 1, sender.count(), "once the parent is accepted, the held reply delivers") + assert.Equal(t, store.DeliveryStateDelivered, getDelivery(t, conn, id).State) +} + +func TestCausalGating_BridgeParentAcceptedIsEligible(t *testing.T) { + conn := workerTestDB(t) + seedWorkerActor(t, conn, true, false) + seedBridgeParent(t, conn, true) // accepted_at set + id := seedDelivery(t, conn, "Create", gParentATURI, createPayload("x")) + + sender := &fakeSender{} + w := newWorker(t, conn, sender, nil) + worked, err := w.DeliverNext(context.Background()) + require.NoError(t, err) + assert.True(t, worked) + + assert.Equal(t, 1, sender.count(), "once the parent is accepted, the reply delivers") + assert.Equal(t, store.DeliveryStateDelivered, getDelivery(t, conn, id).State) +} + +func TestCausalGating_FediverseParentIsAlwaysEligible(t *testing.T) { + conn := workerTestDB(t) + seedWorkerActor(t, conn, true, false) + // NO outbound_objects row for the parent: it is a fediverse-origin object + // (a Lemmy post we never federated outward) — already on Lemmy, so the reply + // is eligible immediately. This is the crux: a naive gate poisons every + // reply-to-a-Lemmy-post. + id := seedDelivery(t, conn, "Create", "at://did:plc:someone/social.coves.community.comment/lemmyparent", + createPayload("x")) + + sender := &fakeSender{} + w := newWorker(t, conn, sender, nil) + worked, err := w.DeliverNext(context.Background()) + require.NoError(t, err) + assert.True(t, worked) + + assert.Equal(t, 1, sender.count(), + "a reply whose parent has no outbound_objects row (fediverse-origin) delivers immediately") + assert.Equal(t, store.DeliveryStateDelivered, getDelivery(t, conn, id).State) +} + +func TestCausalGating_BoundedWaitPoisonsParentUnaccepted(t *testing.T) { + conn := workerTestDB(t) + seedWorkerActor(t, conn, true, false) + seedBridgeParent(t, conn, false) // never accepted + id := seedDelivery(t, conn, "Create", gParentATURI, createPayload("x")) + // The bounded wait is exhausted (attempts at the cap): a comment on a + // never-accepted post must not wait forever. + setAttempts(t, conn, id, 3) // MaxAttempts is 3 + + w := newWorker(t, conn, &fakeSender{}, nil) + _, err := w.DeliverNext(context.Background()) + require.NoError(t, err) + + d := getDelivery(t, conn, id) + assert.Equal(t, store.DeliveryStatePoisoned, d.State, + "after the bounded wait, an unaccepted parent poisons the child") + assert.Contains(t, d.LastErrorClass, "parent_unaccepted", + "the poison reason is queryable: parent_unaccepted (distinct from a delivery failure)") +} + +func TestCausalGating_PoisonedParentPoisonsChild(t *testing.T) { + conn := workerTestDB(t) + seedWorkerActor(t, conn, true, false) + seedBridgeParent(t, conn, false) + + // The parent's own delivery is POISONED. A descendant must not wait forever + // for a parent that will never land — it poisons with a DISTINCT reason. + parentActivityID := "https://coves.social/ap/activity/" + repeatHex64("parent") + _, err := store.NewOutboundActivities(conn).Insert(context.Background(), store.OutboundActivity{ + ActivityID: parentActivityID, + ActorDID: wActorDID, + Kind: "Create", + Payload: createPayload(parentActivityID), + }) + require.NoError(t, err) + _, err = store.NewOutboundDeliveries(conn).Enqueue(context.Background(), store.OutboundDelivery{ + ActivityID: parentActivityID, + TargetInbox: wInbox, + OrderingKey: wCommunityAPID, + }) + require.NoError(t, err) + _, err = conn.ExecContext(context.Background(), + `UPDATE outbound_deliveries SET state='poisoned' WHERE activity_id=$1`, parentActivityID) + require.NoError(t, err) + + id := seedDelivery(t, conn, "Create", gParentATURI, createPayload("child")) + + w := newWorker(t, conn, &fakeSender{}, nil) + _, err = w.DeliverNext(context.Background()) + require.NoError(t, err) + + d := getDelivery(t, conn, id) + assert.Equal(t, store.DeliveryStatePoisoned, d.State, + "a poisoned parent poisons its descendants") + assert.Contains(t, d.LastErrorClass, "parent_poisoned", + "the reason is parent_poisoned — distinct and queryable from parent_unaccepted") +} diff --git a/internal/outbound/inbox_discovery_test.go b/internal/outbound/inbox_discovery_test.go new file mode 100644 index 0000000..f49ca42 --- /dev/null +++ b/internal/outbound/inbox_discovery_test.go @@ -0,0 +1,132 @@ +package outbound + +import ( + "context" + "database/sql" + "net/http" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/ap" + "tidepool/internal/store" +) + +// Task 15 cycle H: inbox discovery + rotation. The target inbox is the +// community Group's endpoints.sharedInbox, TTL-cached. A 401/404/410 on delivery +// triggers ONE cache-bypassing re-resolution before poisoning — an endpoint +// rotation must not become a poison; a still-bad inbox after the re-resolve +// does poison. + +// rotatingResolver serves a stored inbox normally and a DIFFERENT one on the +// cache-bypassing fresh path (the rotation). +type rotatingResolver struct { + normal string + fresh string + freshCalls int +} + +func (r *rotatingResolver) ResolveInbox(context.Context, string) (string, error) { + return r.normal, nil +} + +func (r *rotatingResolver) ResolveInboxFresh(context.Context, string) (string, error) { + r.freshCalls++ + return r.fresh, nil +} + +func deliveryState(t *testing.T, conn *sql.DB, activityID string) store.DeliveryState { + t.Helper() + var s string + require.NoError(t, conn.QueryRowContext(context.Background(), + `SELECT state FROM outbound_deliveries WHERE activity_id = $1`, activityID).Scan(&s)) + return store.DeliveryState(s) +} + +func TestInboxRotation_ReresolveThenDeliver(t *testing.T) { + conn := workerTestDB(t) + seedWorkerActor(t, conn, true, false) + id := seedDelivery(t, conn, "Create", "", createPayload("x")) + + rotated := "https://lemmy.world/c/tech/inbox-v2" + resolver := &rotatingResolver{normal: wInbox, fresh: rotated} + + // The stored inbox 404s; the rotated inbox accepts. + sender := &fakeSender{respond: func(_ int, inbox string) error { + if inbox == rotated { + return nil + } + return httpErr(http.StatusNotFound, "") + }} + + w := newWorker(t, conn, sender, func(o *WorkerOptions) { o.Inboxes = resolver }) + worked, err := w.DeliverNext(context.Background()) + require.NoError(t, err) + assert.True(t, worked) + + assert.Equal(t, 1, resolver.freshCalls, + "a 404 triggers exactly ONE cache-bypassing re-resolution (endpoint rotation, not poison)") + assert.Equal(t, store.DeliveryStateDelivered, deliveryState(t, conn, id), + "after re-resolving to the rotated inbox, the delivery succeeds") +} + +func TestInboxRotation_StillBadAfterReresolvePoisons(t *testing.T) { + conn := workerTestDB(t) + seedWorkerActor(t, conn, true, false) + id := seedDelivery(t, conn, "Create", "", createPayload("x")) + + resolver := &rotatingResolver{normal: wInbox, fresh: "https://lemmy.world/c/tech/inbox-v2"} + // Every inbox 404s — the endpoint is genuinely gone, not rotated. + sender := senderReturning(httpErr(http.StatusNotFound, "")) + + w := newWorker(t, conn, sender, func(o *WorkerOptions) { o.Inboxes = resolver }) + _, err := w.DeliverNext(context.Background()) + require.NoError(t, err) + + assert.Equal(t, 1, resolver.freshCalls, "the re-resolve is attempted exactly once") + assert.Equal(t, store.DeliveryStatePoisoned, deliveryState(t, conn, id), + "a still-404 inbox after the single re-resolve poisons (the endpoint is really gone)") +} + +// --------------------------------------------------------------------------- +// The TTL-cached resolver (NewInboxResolver): a cache hit avoids re-fetching +// the Group doc; the fresh path bypasses it. +// --------------------------------------------------------------------------- + +type countingFetcher struct { + calls int + doc *ap.Object +} + +func (f *countingFetcher) FetchActor(context.Context, string) (*ap.Object, error) { + f.calls++ + return f.doc, nil +} + +func TestInboxResolver_CachesAndBypasses(t *testing.T) { + shared := "https://lemmy.world/c/tech/inbox" + fetcher := &countingFetcher{doc: &ap.Object{ + ID: wCommunityAPID, + Type: "Group", + Inbox: shared, + Endpoints: &ap.Endpoints{SharedInbox: shared}, + }} + resolver := NewInboxResolver(fetcher, time.Minute) + ctx := context.Background() + + inbox, err := resolver.ResolveInbox(ctx, wCommunityAPID) + require.NoError(t, err) + assert.Equal(t, shared, inbox, "the resolver returns endpoints.sharedInbox") + + _, err = resolver.ResolveInbox(ctx, wCommunityAPID) + require.NoError(t, err) + assert.Equal(t, 1, fetcher.calls, "a second resolve within the TTL is served from cache (no re-fetch)") + + fresh, ok := resolver.(FreshInboxResolver) + require.True(t, ok, "the cached resolver must expose the cache-bypassing FreshInboxResolver") + _, err = fresh.ResolveInboxFresh(ctx, wCommunityAPID) + require.NoError(t, err) + assert.Equal(t, 2, fetcher.calls, "the fresh path bypasses the cache and re-fetches the Group doc") +} diff --git a/internal/outbound/inbox_resolver.go b/internal/outbound/inbox_resolver.go new file mode 100644 index 0000000..53ee215 --- /dev/null +++ b/internal/outbound/inbox_resolver.go @@ -0,0 +1,81 @@ +package outbound + +import ( + "context" + "fmt" + "sync" + "time" + + "tidepool/internal/ap" +) + +// ActorFetcher fetches an AP actor document by IRI. *ap.Client satisfies it via +// FetchActor (SSRF-guarded, same-authority binding); the resolver depends only +// on this narrow surface. +type ActorFetcher interface { + FetchActor(ctx context.Context, iri string) (*ap.Object, error) +} + +// cachedInboxResolver resolves a community's target inbox from its Group actor +// document (preferring endpoints.sharedInbox), memoized for a TTL. It also +// implements FreshInboxResolver so the worker can bypass the cache once on an +// endpoint rotation before poisoning. +type cachedInboxResolver struct { + fetcher ActorFetcher + ttl time.Duration + + mu sync.Mutex + cache map[string]inboxEntry +} + +type inboxEntry struct { + inbox string + expires time.Time +} + +// NewInboxResolver builds the TTL-cached inbox resolver. +func NewInboxResolver(fetcher ActorFetcher, ttl time.Duration) InboxResolver { + return &cachedInboxResolver{ + fetcher: fetcher, + ttl: ttl, + cache: make(map[string]inboxEntry), + } +} + +// ResolveInbox returns the community's inbox, from cache when fresh. +func (r *cachedInboxResolver) ResolveInbox(ctx context.Context, communityAPID string) (string, error) { + r.mu.Lock() + entry, ok := r.cache[communityAPID] + r.mu.Unlock() + if ok && time.Now().Before(entry.expires) { + return entry.inbox, nil + } + return r.fetchAndCache(ctx, communityAPID) +} + +// ResolveInboxFresh re-fetches the Group doc bypassing the cache (rotation): an +// endpoint rotation must not become a poison, so the worker asks for a fresh +// resolution ONCE on a 401/404/410 before giving up. +func (r *cachedInboxResolver) ResolveInboxFresh(ctx context.Context, communityAPID string) (string, error) { + return r.fetchAndCache(ctx, communityAPID) +} + +// fetchAndCache fetches the Group document and reads its delivery inbox. The +// fetch runs through the ap client's SSRF + same-authority guards (ActorFetcher +// is *ap.Client.FetchActor), so a Group doc advertising a cross-authority or +// private-range inbox is refused at fetch time; the worker's POST re-applies the +// egress guard on the resolved inbox. +func (r *cachedInboxResolver) fetchAndCache(ctx context.Context, communityAPID string) (string, error) { + doc, err := r.fetcher.FetchActor(ctx, communityAPID) + if err != nil { + return "", fmt.Errorf("resolve inbox for %s: %w", communityAPID, err) + } + inbox := doc.SharedInboxOrInbox() + if inbox == "" { + return "", fmt.Errorf("community %s advertises no inbox", communityAPID) + } + r.mu.Lock() + r.cache[communityAPID] = inboxEntry{inbox: inbox, expires: time.Now().Add(r.ttl)} + r.mu.Unlock() + return inbox, nil +} diff --git a/internal/outbound/outbound.go b/internal/outbound/outbound.go index 6f20c47..ddcfebf 100644 --- a/internal/outbound/outbound.go +++ b/internal/outbound/outbound.go @@ -48,7 +48,52 @@ type InboxResolver interface { // ActivitySender POSTs a signed activity to a remote inbox as a specific // persona. *ap.Client satisfies it via SendActivityAs (the per-actor signing -// path, distinct from the service actor's configured-Signer SendActivity). +// path, distinct from the service actor's configured-Signer SendActivity). A +// non-2xx response is returned as an *ap.HTTPError carrying the status and a +// bounded body excerpt, so the worker can classify Lemmy's duplicate-activity +// response (400 + "already received") as DELIVERED; a transport failure is any +// other error. type ActivitySender interface { SendActivityAs(ctx context.Context, signer *ap.Signer, inbox string, activity any) error } + +// FreshInboxResolver is the OPTIONAL cache-bypassing extension an InboxResolver +// may implement. On a 401/404/410 the worker re-resolves the community's inbox +// ONCE bypassing the TTL cache — an endpoint rotation must not become a poison. +// A resolver that does not implement it is simply asked again through the +// normal (cached) path. +type FreshInboxResolver interface { + ResolveInboxFresh(ctx context.Context, communityAPID string) (inbox string, err error) +} + +// DeliveryScope is the (actor, community, inbox host) a delivery falls under, +// which the kill switches are keyed on. +type DeliveryScope struct { + ActorDID string + CommunityAPID string + InboxHost string +} + +// Switches is the outbound kill-switch surface (decision 19), consulted at +// claim time. A delivery a switch blocks is PARKED — it stays pending and +// resumes when the switch clears — never poisoned or cancelled. The four levels +// (global, per-host, per-community, per-actor) all funnel through +// OutboundAllowed. DryRun is the separate "translate and log but POST nothing" +// mode. +type Switches interface { + // OutboundAllowed reports whether a delivery in this scope may be sent. + OutboundAllowed(scope DeliveryScope) bool + // DryRun reports whether to translate + log without POSTing (the delivery + // stays pending, nothing is marked delivered). + DryRun() bool +} + +// AllowAll is the default Switches: everything enabled, no dry-run. main wires a +// config-backed implementation; a nil Switches on the Worker means AllowAll. +type AllowAll struct{} + +// OutboundAllowed always allows. +func (AllowAll) OutboundAllowed(DeliveryScope) bool { return true } + +// DryRun is always false. +func (AllowAll) DryRun() bool { return false } diff --git a/internal/outbound/worker.go b/internal/outbound/worker.go index 619bdf3..fd5b96e 100644 --- a/internal/outbound/worker.go +++ b/internal/outbound/worker.go @@ -3,9 +3,17 @@ package outbound import ( "context" "database/sql" + "encoding/json" + stderrors "errors" + "fmt" "log/slog" + "net/http" + "net/url" + "strings" "time" + "tidepool/internal/ap" + "tidepool/internal/errors" "tidepool/internal/store" ) @@ -27,33 +35,64 @@ type WorkerOptions struct { // create/update is cancelled, not delivered (delete/undo are exempt). // Optional: nil from DB. Actors store.APActors + // Prefs is the federation opt-out half of the consent recheck (enabled=false + // → an outward delivery is cancelled). Optional: nil from DB. + Prefs store.FederationPrefs + // Votes receives the delivery-success callbacks (decision 16): a Like/Dislike + // success flips delivered_state; an Undo success clears the row. Optional: + // nil from DB. + Votes store.OutboundVotes // Signers yields the per-actor Signer each delivery is signed with. Signers SignerProvider // Inboxes re-resolves a rotated inbox once before poisoning. Inboxes InboxResolver // Sender POSTs the signed activity. *ap.Client satisfies it. Sender ActivitySender + // Switches are the outbound kill switches + dry-run (decision 19). Nil means + // AllowAll (everything enabled, no dry-run). + Switches Switches + // MaxAttempts caps delivery attempts before a retryable failure poisons. + // Zero uses DefaultMaxDeliveryAttempts. + MaxAttempts int + // BackoffBase is the first retry-backoff step (doubles per attempt). Zero + // uses DefaultBackoffBase; tests compress it. + BackoffBase time.Duration // Lease overrides DefaultLease. Lease time.Duration // Logger receives per-delivery outcomes. Nil uses slog.Default(). Logger *slog.Logger } +// Delivery retry bounds. +const ( + // DefaultMaxDeliveryAttempts caps attempts before a retryable failure + // (transport/5xx/429) or an unaccepted parent poisons. + DefaultMaxDeliveryAttempts = 8 + // DefaultBackoffBase is the first retry step; it doubles per attempt, + // capped at one hour. + DefaultBackoffBase = 30 * time.Second +) + // Worker claims one delivery at a time and carries it to a terminal state. It // is the at-least-once engine: a crash between POST and MarkDelivered redelivers // (Lemmy dedupes on our stable activity id, and its duplicate-activity response // is classified DELIVERED, not poisoned). type Worker struct { - db *sql.DB - activities store.OutboundActivities - deliveries store.OutboundDeliveries - objects store.OutboundObjects - actors store.APActors - signers SignerProvider - inboxes InboxResolver - sender ActivitySender - lease time.Duration - logger *slog.Logger + db *sql.DB + activities store.OutboundActivities + deliveries store.OutboundDeliveries + objects store.OutboundObjects + actors store.APActors + prefs store.FederationPrefs + votes store.OutboundVotes + signers SignerProvider + inboxes InboxResolver + sender ActivitySender + switches Switches + maxAttempts int + backoffBase time.Duration + lease time.Duration + logger *slog.Logger } // NewWorker wires a Worker. @@ -78,27 +117,465 @@ func NewWorker(opts WorkerOptions) (*Worker, error) { if objects == nil { objects = store.NewOutboundObjects(opts.DB) } + prefs := opts.Prefs + if prefs == nil { + prefs = store.NewFederationPrefs(opts.DB) + } + votes := opts.Votes + if votes == nil { + votes = store.NewOutboundVotes(opts.DB) + } + var switches Switches = opts.Switches + if switches == nil { + switches = AllowAll{} + } + maxAttempts := opts.MaxAttempts + if maxAttempts <= 0 { + maxAttempts = DefaultMaxDeliveryAttempts + } + backoffBase := opts.BackoffBase + if backoffBase <= 0 { + backoffBase = DefaultBackoffBase + } return &Worker{ - db: opts.DB, - activities: activities, - deliveries: deliveries, - objects: objects, - actors: opts.Actors, - signers: opts.Signers, - inboxes: opts.Inboxes, - sender: opts.Sender, - lease: lease, - logger: logger, + db: opts.DB, + activities: activities, + deliveries: deliveries, + objects: objects, + actors: opts.Actors, + prefs: prefs, + votes: votes, + signers: opts.Signers, + inboxes: opts.Inboxes, + sender: opts.Sender, + switches: switches, + maxAttempts: maxAttempts, + backoffBase: backoffBase, + lease: lease, + logger: logger, }, nil } +// Run drives DeliverNext in a loop until ctx is cancelled, sleeping idle when +// the queue drains. A per-delivery error is logged and the loop continues — one +// bad delivery must not stop the pipe. +func (w *Worker) Run(ctx context.Context, idle time.Duration) error { + for { + if err := ctx.Err(); err != nil { + return err + } + worked, err := w.DeliverNext(ctx) + if err != nil { + w.logger.Error("outbound delivery failed", "error", err) + } + if !worked { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(idle): + } + } + } +} + // DeliverNext claims one processable delivery and carries it to a terminal -// state (delivered / poisoned / cancelled) or reschedules it. It returns -// worked=true when a delivery was claimed and handled, worked=false (nil error) -// when the queue held nothing claimable. -// -// STUB (task 15 RED): GREEN claims, rechecks consent (exempting delete/undo), -// enforces causal gating, signs and POSTs, then records the fenced outcome. +// state (delivered / poisoned / cancelled), parks it (kill switch / dry-run / +// causal wait), or reschedules it. It returns worked=true when a delivery was +// claimed and handled, worked=false (nil error) when the queue held nothing +// claimable. Only an infrastructure failure returns a non-nil error; every +// normal delivery outcome is a handled success. func (w *Worker) DeliverNext(ctx context.Context) (worked bool, err error) { - return false, nil + delivery, err := w.deliveries.ClaimNext(ctx, w.lease) + if err != nil { + if errors.IsNotFound(err) { + return false, nil // empty queue + } + return false, fmt.Errorf("claim next delivery: %w", err) + } + if err := w.handle(ctx, delivery); err != nil { + return true, err + } + return true, nil +} + +// handle carries one claimed delivery to its outcome under its fencing token. +func (w *Worker) handle(ctx context.Context, delivery *store.OutboundDelivery) error { + activity, err := w.activities.Get(ctx, delivery.ActivityID) + if err != nil { + return fmt.Errorf("load activity %s: %w", delivery.ActivityID, err) + } + + // Kill switch (decision 19): an operator block PARKS the delivery — it stays + // pending and resumes when the switch clears, never poisoned or cancelled. + scope := DeliveryScope{ + ActorDID: activity.ActorDID, + CommunityAPID: delivery.OrderingKey, + InboxHost: hostOf(delivery.TargetInbox), + } + if !w.switches.OutboundAllowed(scope) { + return w.park(ctx, delivery, "switch_parked", "outbound kill switch engaged") + } + if w.switches.DryRun() { + w.logger.Info("dry-run: delivery translated but not POSTed", + "activity", delivery.ActivityID, "inbox", delivery.TargetInbox) + return w.park(ctx, delivery, "dry_run", "dry-run mode") + } + + // Causal gate (decision 15): a reply must not be delivered before its + // bridge-origin parent is accepted. Checked POST-claim on the row (not a + // join in ClaimNext) so the loose-index-scan is untouched. + switch w.causalStatus(ctx, delivery, activity) { + case causalEligible: + // fall through to consent + delivery + case causalWait: + return w.park(ctx, delivery, "parent_pending", "waiting for bridge-origin parent to be accepted") + case causalPoisonUnaccepted: + return w.poison(ctx, delivery, "parent_unaccepted", "bounded wait exhausted; parent never accepted", 0) + case causalPoisonParent: + return w.poison(ctx, delivery, "parent_poisoned", "parent delivery poisoned; descendant cannot land", 0) + } + + // Consent recheck (retraction asymmetry): a Delete/Undo always goes out — + // it is how an opted-out user takes down what is already federated. Outward + // kinds are cancelled when the actor is disabled, paused, or opted out. + if !isRetraction(activity.Kind) { + blocked, err := w.consentBlocked(ctx, activity.ActorDID) + if err != nil { + return err + } + if blocked { + if _, err := w.deliveries.CancelForActor(ctx, activity.ActorDID); err != nil { + return fmt.Errorf("cancel deliveries for %s: %w", activity.ActorDID, err) + } + return nil + } + } + + return w.deliver(ctx, delivery, activity) +} + +// deliver signs and POSTs the stored payload verbatim, then classifies the +// outcome. +func (w *Worker) deliver(ctx context.Context, delivery *store.OutboundDelivery, activity *store.OutboundActivity) error { + signer, err := w.signers.SignerFor(ctx, activity.ActorDID) + if err != nil { + // A signer that cannot be resolved right now is transient (a KEK blip, + // a not-yet-replicated actor): retry rather than poison. + return w.releaseOrPoison(ctx, delivery, "signer", err.Error(), 0) + } + + err = w.sender.SendActivityAs(ctx, signer, delivery.TargetInbox, json.RawMessage(activity.Payload)) + return w.classify(ctx, delivery, activity, signer, err) +} + +// classify maps a POST outcome onto the retry taxonomy. +func (w *Worker) classify(ctx context.Context, delivery *store.OutboundDelivery, activity *store.OutboundActivity, signer *ap.Signer, err error) error { + if err == nil { + return w.deliverSuccess(ctx, delivery, activity, http.StatusAccepted) + } + + var he ap.HTTPError + if stderrors.As(err, &he) { + switch { + case isDuplicate(he): + // Lemmy's received_activity dedupe (400 + "already received") is a + // SUCCESS by our stable id: a redelivery after a crash is expected. + return w.deliverSuccess(ctx, delivery, activity, he.StatusCode) + case he.StatusCode == http.StatusUnauthorized || + he.StatusCode == http.StatusNotFound || + he.StatusCode == http.StatusGone: + return w.rotateInbox(ctx, delivery, activity, signer, he) + case he.StatusCode == http.StatusRequestTimeout || + he.StatusCode == http.StatusTooManyRequests || + he.StatusCode >= 500: + return w.releaseOrPoison(ctx, delivery, classForStatus(he.StatusCode), he.Body, he.StatusCode) + default: + // Other 4xx: a genuine rejection. Retried on a small budget, then + // poisoned. + return w.releaseOrPoison(ctx, delivery, "4xx", he.Body, he.StatusCode) + } + } + // Transport failure (dial/TLS/timeout): transient. + return w.releaseOrPoison(ctx, delivery, "transport", err.Error(), 0) +} + +// rotateInbox handles a 401/404/410: re-resolve the community's inbox ONCE +// bypassing the cache (an endpoint rotation must not become a poison), retry, +// then deliver-or-poison. +func (w *Worker) rotateInbox(ctx context.Context, delivery *store.OutboundDelivery, activity *store.OutboundActivity, signer *ap.Signer, first ap.HTTPError) error { + fresh, ok := w.inboxes.(FreshInboxResolver) + if !ok { + return w.releaseOrPoison(ctx, delivery, "inbox_gone", first.Body, first.StatusCode) + } + inbox, err := fresh.ResolveInboxFresh(ctx, delivery.OrderingKey) + if err != nil { + return w.releaseOrPoison(ctx, delivery, "inbox_resolve", err.Error(), first.StatusCode) + } + + err = w.sender.SendActivityAs(ctx, signer, inbox, json.RawMessage(activity.Payload)) + if err == nil { + return w.deliverSuccess(ctx, delivery, activity, http.StatusAccepted) + } + var he ap.HTTPError + if stderrors.As(err, &he) && isDuplicate(he) { + return w.deliverSuccess(ctx, delivery, activity, he.StatusCode) + } + // Still bad after the single re-resolve: the endpoint is genuinely gone. + status := first.StatusCode + if stderrors.As(err, &he) { + status = he.StatusCode + } + return w.poison(ctx, delivery, "inbox_gone", "inbox still unreachable after re-resolution", status) +} + +// deliverSuccess marks the delivery delivered under its fencing token and fires +// the object-acceptance and vote-delivery callbacks. +func (w *Worker) deliverSuccess(ctx context.Context, delivery *store.OutboundDelivery, activity *store.OutboundActivity, status int) error { + _, applied, err := w.deliveries.MarkDelivered(ctx, delivery.ActivityID, delivery.TargetInbox, status, *delivery.ClaimedUntil) + if err != nil { + return fmt.Errorf("mark delivered %s: %w", delivery.ActivityID, err) + } + if !applied { + return nil // a stale claim: another worker already recorded the outcome + } + w.stampAccepted(ctx, activity) + return w.voteCallback(ctx, activity) +} + +// stampAccepted opens the causal gate for this object's children: on a +// successful Create/Update, the object it federated is now accepted by its +// community. Best-effort — a delivery for an object with no outbound_objects row +// (a comment we never persisted, a vote) simply has nothing to stamp. +func (w *Worker) stampAccepted(ctx context.Context, activity *store.OutboundActivity) { + if activity.Kind != "Create" && activity.Kind != "Update" { + return + } + atURI := objectATURIFromPayload(activity.Payload) + if atURI == "" { + return + } + if err := w.objects.SetAccepted(ctx, atURI); err != nil && !errors.IsNotFound(err) { + w.logger.Warn("stamp accepted failed", "at_uri", atURI, "error", err) + } +} + +// voteCallback applies decision-16 delivery callbacks: a Like/Dislike success +// flips outbound_votes.delivered_state; an Undo success clears the row. +func (w *Worker) voteCallback(ctx context.Context, activity *store.OutboundActivity) error { + switch activity.Kind { + case "Like", "Dislike": + vote, err := w.votes.GetByActivityID(ctx, activity.ActivityID) + if errors.IsNotFound(err) { + return nil + } + if err != nil { + return fmt.Errorf("resolve vote for %s: %w", activity.ActivityID, err) + } + if err := w.votes.SetDeliveredState(ctx, vote.VoteATURI, store.DeliveredStateDelivered); err != nil { + return fmt.Errorf("flip vote %s delivered: %w", vote.VoteATURI, err) + } + case "Undo": + innerID := innerObjectID(activity.Payload) + if innerID == "" { + return nil + } + vote, err := w.votes.GetByActivityID(ctx, innerID) + if errors.IsNotFound(err) { + return nil + } + if err != nil { + return fmt.Errorf("resolve vote for undo %s: %w", innerID, err) + } + if err := w.votes.Delete(ctx, vote.VoteATURI); err != nil { + return fmt.Errorf("clear vote %s on undo: %w", vote.VoteATURI, err) + } + } + return nil +} + +// releaseOrPoison reschedules a retryable failure with backoff, or poisons it +// once the attempt cap is reached (the claim already bumped attempts). +func (w *Worker) releaseOrPoison(ctx context.Context, delivery *store.OutboundDelivery, class, excerpt string, status int) error { + if delivery.Attempts >= w.maxAttempts { + return w.poison(ctx, delivery, class, excerpt, status) + } + next := time.Now().Add(w.backoff(delivery.Attempts)) + _, _, err := w.deliveries.Release(ctx, delivery.ActivityID, delivery.TargetInbox, class, excerpt, status, next, *delivery.ClaimedUntil) + if err != nil { + return fmt.Errorf("release delivery %s: %w", delivery.ActivityID, err) + } + return nil +} + +// poison marks the delivery permanently failed under its fencing token. +func (w *Worker) poison(ctx context.Context, delivery *store.OutboundDelivery, class, excerpt string, status int) error { + _, _, err := w.deliveries.MarkPoisoned(ctx, delivery.ActivityID, delivery.TargetInbox, class, excerpt, status, *delivery.ClaimedUntil) + if err != nil { + return fmt.Errorf("poison delivery %s: %w", delivery.ActivityID, err) + } + return nil +} + +// park releases the delivery pending with a short backoff and no move toward +// the poison cap: a kill switch, dry-run, or causal wait is a "not now", never +// a failure. +func (w *Worker) park(ctx context.Context, delivery *store.OutboundDelivery, class, reason string) error { + _, _, err := w.deliveries.Release(ctx, delivery.ActivityID, delivery.TargetInbox, class, reason, 0, time.Now(), *delivery.ClaimedUntil) + if err != nil { + return fmt.Errorf("park delivery %s: %w", delivery.ActivityID, err) + } + return nil +} + +// causalStatus classifies a delivery's causal eligibility. +type causalStatus int + +const ( + causalEligible causalStatus = iota + causalWait + causalPoisonUnaccepted + causalPoisonParent +) + +func (w *Worker) causalStatus(ctx context.Context, delivery *store.OutboundDelivery, activity *store.OutboundActivity) causalStatus { + if activity.ParentATURI == "" { + return causalEligible + } + parent, err := w.objects.GetByATURI(ctx, activity.ParentATURI) + if errors.IsNotFound(err) { + // No outbound_objects row: the parent is a FEDIVERSE-origin object + // (already on the peer), so the reply is always eligible. This is the + // crux — a naive gate would poison every reply to a Lemmy post. + return causalEligible + } + if err != nil { + w.logger.Error("causal parent lookup failed", "parent", activity.ParentATURI, "error", err) + return causalWait // transient: hold rather than poison on a lookup blip + } + if parent.IsAccepted() { + return causalEligible + } + // Bridge-origin parent, not yet accepted. A poisoned ancestor on the same + // serial line means it will NEVER land → poison the descendant distinctly. + poisoned, err := w.deliveries.HasPoisonedPredecessor(ctx, delivery.OrderingKey, delivery.TargetInbox, delivery.Seq) + if err != nil { + w.logger.Error("poisoned-predecessor check failed", "error", err) + return causalWait + } + if poisoned { + return causalPoisonParent + } + if delivery.Attempts >= w.maxAttempts { + return causalPoisonUnaccepted // bounded wait exhausted + } + return causalWait +} + +// consentBlocked reports whether an outward delivery for actorDID must be +// cancelled: the actor is disabled, delivery-paused, or has a federation +// opt-out. A missing pref MEANS default-on (not blocked). +func (w *Worker) consentBlocked(ctx context.Context, actorDID string) (bool, error) { + actor, err := w.actors.GetByDID(ctx, actorDID) + if err != nil { + if errors.IsNotFound(err) { + return true, nil // no actor to sign as: cannot deliver + } + return false, fmt.Errorf("load actor %s: %w", actorDID, err) + } + if !actor.Enabled || actor.DeliveryPaused { + return true, nil + } + pref, err := w.prefs.Get(ctx, actorDID) + if errors.IsNotFound(err) { + return false, nil // default-on + } + if err != nil { + return false, fmt.Errorf("load federation pref %s: %w", actorDID, err) + } + return !pref.Enabled, nil +} + +// backoff is the retry delay for a given attempt count: backoffBase doubled per +// attempt, capped at one hour. +func (w *Worker) backoff(attempts int) time.Duration { + const maxBackoff = time.Hour + d := w.backoffBase + for i := 1; i < attempts && d < maxBackoff; i++ { + d *= 2 + } + if d > maxBackoff { + d = maxBackoff + } + return d +} + +// isRetraction reports whether a kind is a take-down that is exempt from the +// consent recheck (a Delete or a vote Undo). +func isRetraction(kind string) bool { return kind == "Delete" || kind == "Undo" } + +// isDuplicate reports whether an HTTPError is Lemmy's duplicate-activity +// response (a 400 whose body reports the activity was already received). +func isDuplicate(he ap.HTTPError) bool { + return he.StatusCode == http.StatusBadRequest && strings.Contains(strings.ToLower(he.Body), "already") +} + +// classForStatus labels a retryable HTTP status for the retry taxonomy. +func classForStatus(status int) string { + switch { + case status == http.StatusRequestTimeout: + return "timeout" + case status == http.StatusTooManyRequests: + return "rate_limited" + case status >= 500: + return "5xx" + default: + return "transient" + } +} + +// hostOf returns the lowercase host of a URL, or "" if it does not parse. +func hostOf(rawURL string) string { + parsed, err := url.Parse(rawURL) + if err != nil { + return "" + } + return strings.ToLower(parsed.Hostname()) +} + +// objectATURIFromPayload recovers the at-uri of the object a Create/Update +// federated, from the object's id URL (/ap/object/{did}/{collection}/{rkey}), so +// its acceptance can be stamped. Returns "" when the payload carries no such id. +func objectATURIFromPayload(payload []byte) string { + var activity struct { + Object struct { + ID string `json:"id"` + } `json:"object"` + } + if err := json.Unmarshal(payload, &activity); err != nil { + return "" + } + const marker = "/ap/object/" + idx := strings.Index(activity.Object.ID, marker) + if idx < 0 { + return "" + } + suffix := activity.Object.ID[idx+len(marker):] + if suffix == "" { + return "" + } + return "at://" + suffix +} + +// innerObjectID reads the embedded inner object's id from an Undo payload (the +// Like/Dislike activity id the Undo withdraws), which resolves the vote row. +func innerObjectID(payload []byte) string { + var activity struct { + Object struct { + ID string `json:"id"` + } `json:"object"` + } + if err := json.Unmarshal(payload, &activity); err != nil { + return "" + } + return activity.Object.ID } diff --git a/internal/outbound/worker_test.go b/internal/outbound/worker_test.go new file mode 100644 index 0000000..7ad8d80 --- /dev/null +++ b/internal/outbound/worker_test.go @@ -0,0 +1,518 @@ +package outbound + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "fmt" + "net/http" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/ap" + "tidepool/internal/errors" + "tidepool/internal/store" + "tidepool/internal/testutil" +) + +// Task 15 cycle F: the delivery Worker. DeliverNext claims a delivery, rechecks +// consent, resolves the signer, POSTs the STORED activity payload verbatim, and +// records the fenced outcome. The retry taxonomy, the consent/retraction +// asymmetry, and the vote-delivery callbacks are pinned here; the signed-wire + +// addressing + verify-against-served-doc happy path is the OUTER test. + +const ( + wActorDID = "did:plc:workeractor000000000000" + wActorID = "https://coves.social/ap/actor/" + wActorDID + wCommunityDID = "did:plc:44ybard66vv44zksje25o7dz" + wCommunityAPID = "https://lemmy.world/c/tech" + wInbox = "https://lemmy.world/c/tech/inbox" +) + +func workerTestDB(t *testing.T) *sql.DB { + t.Helper() + database := testutil.DB(t) + testutil.Truncate(t, database, + "outbound_deliveries", "outbound_activities", "outbound_objects", + "outbound_votes", "ap_actors", "federation_prefs") + return database +} + +// ---- fakes ---- + +type sentPost struct { + inbox string + payload []byte +} + +// fakeSender records POSTs and returns a scripted outcome per call. +type fakeSender struct { + mu sync.Mutex + posts []sentPost + respond func(callNo int, inbox string) error // nil => 2xx success +} + +func (s *fakeSender) SendActivityAs(_ context.Context, _ *ap.Signer, inbox string, activity any) error { + payload, _ := json.Marshal(activity) + s.mu.Lock() + s.posts = append(s.posts, sentPost{inbox: inbox, payload: payload}) + call := len(s.posts) + respond := s.respond + s.mu.Unlock() + if respond != nil { + return respond(call, inbox) + } + return nil +} + +func (s *fakeSender) count() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.posts) +} + +// alwaysFail returns a sender whose every POST returns err. +func senderReturning(err error) *fakeSender { + return &fakeSender{respond: func(int, string) error { return err }} +} + +// httpErr builds the *ap.HTTPError a delivery POST surfaces on a non-2xx +// response (status + bounded body excerpt). +func httpErr(status int, body string) error { + return ap.HTTPError{URL: wInbox, StatusCode: status, Body: body} +} + +// fakeSigners returns one generated signer for any DID. +type fakeSigners struct { + signer *ap.Signer + err error +} + +func newFakeSigners(t *testing.T) fakeSigners { + t.Helper() + key, err := ap.GenerateRSAKey() + require.NoError(t, err) + return fakeSigners{signer: ap.NewSigner(wActorID+"#main-key", key)} +} + +func (f fakeSigners) SignerFor(context.Context, string) (*ap.Signer, error) { + return f.signer, f.err +} + +// fakeSwitches records the scopes it was consulted with. +type fakeSwitches struct { + mu sync.Mutex + allow bool + dryRun bool + scopes []DeliveryScope +} + +func (s *fakeSwitches) OutboundAllowed(scope DeliveryScope) bool { + s.mu.Lock() + s.scopes = append(s.scopes, scope) + s.mu.Unlock() + return s.allow +} + +func (s *fakeSwitches) DryRun() bool { return s.dryRun } + +// staticInbox always resolves to wInbox. +type staticInbox struct{ inbox string } + +func (r staticInbox) ResolveInbox(context.Context, string) (string, error) { return r.inbox, nil } + +// ---- seed helpers ---- + +func seedWorkerActor(t *testing.T, conn *sql.DB, enabled, paused bool) { + t.Helper() + ctx := context.Background() + actors := store.NewAPActors(conn) + _, err := actors.Create(ctx, store.APActor{ + DID: wActorDID, + Kind: store.ActorTypePerson, + ActorID: wActorID, + NormalizedOrigin: "coves.social", + LocalPart: "worker", + RSAKeySealed: []byte("sealed"), + RSAKeyVersion: 1, + PublicKeyPEM: "pem", + }) + require.NoError(t, err) + if !enabled { + require.NoError(t, actors.SetEnabled(ctx, wActorDID, false)) + } + if paused { + require.NoError(t, actors.SetPaused(ctx, wActorDID, true)) + } +} + +// seedDelivery inserts an activity + its single pending delivery and returns the +// activity id. +func seedDelivery(t *testing.T, conn *sql.DB, kind, parentATURI string, payload []byte) string { + t.Helper() + ctx := context.Background() + activityID := "https://coves.social/ap/activity/" + repeatHex64(kind) + inserted, err := store.NewOutboundActivities(conn).Insert(ctx, store.OutboundActivity{ + ActivityID: activityID, + ActorDID: wActorDID, + Kind: kind, + Payload: payload, + ParentATURI: parentATURI, + }) + require.NoError(t, err) + require.True(t, inserted) + _, err = store.NewOutboundDeliveries(conn).Enqueue(ctx, store.OutboundDelivery{ + ActivityID: activityID, + TargetInbox: wInbox, + OrderingKey: wCommunityAPID, + }) + require.NoError(t, err) + return activityID +} + +func repeatHex64(seed string) string { + // A deterministic 64-hex digest of the seed's CONTENT (not its length), so + // distinct seeds get distinct activity ids and the SAME seed reproduces the + // same id (the vote tests rely on repeatHex64("Like") matching in two + // places). sha256 is exactly 32 bytes → 64 hex chars. + sum := sha256.Sum256([]byte(seed)) + return hex.EncodeToString(sum[:]) +} + +func createPayload(activityID string) []byte { + return []byte(fmt.Sprintf(`{"@context":"https://www.w3.org/ns/activitystreams",`+ + `"id":%q,"type":"Create","actor":%q,"object":{"type":"Note","content":"hi"}}`, + activityID, wActorID)) +} + +func setAttempts(t *testing.T, conn *sql.DB, activityID string, n int) { + t.Helper() + _, err := conn.ExecContext(context.Background(), + `UPDATE outbound_deliveries SET attempts = $2 WHERE activity_id = $1`, activityID, n) + require.NoError(t, err) +} + +func getDelivery(t *testing.T, conn *sql.DB, activityID string) *store.OutboundDelivery { + t.Helper() + d, err := store.NewOutboundDeliveries(conn).Get(context.Background(), activityID, wInbox) + require.NoError(t, err) + require.NotNil(t, d) + return d +} + +func newWorker(t *testing.T, conn *sql.DB, sender ActivitySender, opts func(*WorkerOptions)) *Worker { + t.Helper() + o := WorkerOptions{ + DB: conn, + Actors: store.NewAPActors(conn), + Signers: newFakeSigners(t), + Inboxes: staticInbox{inbox: wInbox}, + Sender: sender, + Lease: time.Minute, + MaxAttempts: 3, + BackoffBase: time.Millisecond, + } + if opts != nil { + opts(&o) + } + w, err := NewWorker(o) + require.NoError(t, err) + return w +} + +// --------------------------------------------------------------------------- +// F: retry taxonomy +// --------------------------------------------------------------------------- + +func TestWorker_DuplicateActivityResponseIsDelivered(t *testing.T) { + conn := workerTestDB(t) + seedWorkerActor(t, conn, true, false) + id := seedDelivery(t, conn, "Create", "", createPayload("x")) + + // The fake Lemmy in the OUTER test returns 400 + {"error":"activity was + // already received"} on a duplicate. THIS is the shape the worker keys on. + sender := senderReturning(httpErr(http.StatusBadRequest, `{"error":"activity was already received"}`)) + w := newWorker(t, conn, sender, nil) + + worked, err := w.DeliverNext(context.Background()) + require.NoError(t, err) + assert.True(t, worked) + assert.Equal(t, store.DeliveryStateDelivered, getDelivery(t, conn, id).State, + "a 400 whose body reports an already-received activity is DELIVERED, never poisoned — "+ + "redelivery after a crash is expected and safe") +} + +func TestWorker_TransientFailuresRelease(t *testing.T) { + for _, tc := range []struct { + name string + err error + }{ + {"transport", fmt.Errorf("dial tcp 1.2.3.4:443: connect: connection refused")}, + {"408", httpErr(http.StatusRequestTimeout, "")}, + {"429", httpErr(http.StatusTooManyRequests, "")}, + {"500", httpErr(http.StatusInternalServerError, "")}, + {"503", httpErr(http.StatusServiceUnavailable, "boom")}, + } { + t.Run(tc.name, func(t *testing.T) { + conn := workerTestDB(t) + seedWorkerActor(t, conn, true, false) + id := seedDelivery(t, conn, "Create", "", createPayload("x")) + + w := newWorker(t, conn, senderReturning(tc.err), nil) + worked, err := w.DeliverNext(context.Background()) + require.NoError(t, err) + assert.True(t, worked) + + d := getDelivery(t, conn, id) + assert.Equal(t, store.DeliveryStatePending, d.State, + "a transport error / 408 / 429 / 5xx is a RETRY: the delivery stays pending") + assert.NotEmpty(t, d.LastErrorClass, "the retry taxonomy label is recorded") + assert.True(t, d.NextAttemptAt.After(time.Now().Add(-time.Second)), + "next_attempt_at is advanced for the backoff") + }) + } +} + +func TestWorker_AttemptCapPoisons(t *testing.T) { + conn := workerTestDB(t) + seedWorkerActor(t, conn, true, false) + id := seedDelivery(t, conn, "Create", "", createPayload("x")) + // One attempt below the cap; ClaimNext bumps to the cap, and the persistent + // 5xx then poisons rather than releasing forever. + setAttempts(t, conn, id, 2) // MaxAttempts is 3 + + w := newWorker(t, conn, senderReturning(httpErr(http.StatusBadGateway, "")), nil) + worked, err := w.DeliverNext(context.Background()) + require.NoError(t, err) + assert.True(t, worked) + assert.Equal(t, store.DeliveryStatePoisoned, getDelivery(t, conn, id).State, + "a retryable failure at the attempt cap poisons") +} + +func TestWorker_OtherClientErrorPoisons(t *testing.T) { + conn := workerTestDB(t) + seedWorkerActor(t, conn, true, false) + id := seedDelivery(t, conn, "Create", "", createPayload("x")) + setAttempts(t, conn, id, 2) // small fixed retry budget exhausted + + // A non-duplicate 4xx (a genuine rejection) poisons after the small budget. + w := newWorker(t, conn, senderReturning(httpErr(http.StatusBadRequest, `{"error":"invalid_object"}`)), nil) + worked, err := w.DeliverNext(context.Background()) + require.NoError(t, err) + assert.True(t, worked) + + d := getDelivery(t, conn, id) + assert.Equal(t, store.DeliveryStatePoisoned, d.State, + "a non-duplicate 4xx is a permanent rejection → poisoned") + require.NotNil(t, d.LastStatusCode) + assert.Equal(t, http.StatusBadRequest, *d.LastStatusCode) +} + +// --------------------------------------------------------------------------- +// F: consent recheck at claim (retraction asymmetry) +// --------------------------------------------------------------------------- + +func TestWorker_ConsentAsymmetry(t *testing.T) { + cases := []struct { + name string + disable func(t *testing.T, conn *sql.DB) + kind string + wantState store.DeliveryState + wantPosted bool + explanation string + }{ + { + name: "disabled actor cancels a create", + disable: func(t *testing.T, conn *sql.DB) { seedWorkerActor(t, conn, false, false) }, + kind: "Create", + wantState: store.DeliveryStateCancelled, + wantPosted: false, + explanation: "a disabled actor's create is cancelled, not delivered and not poisoned", + }, + { + name: "paused actor cancels a create", + disable: func(t *testing.T, conn *sql.DB) { seedWorkerActor(t, conn, true, true) }, + kind: "Create", + wantState: store.DeliveryStateCancelled, + wantPosted: false, + explanation: "a delivery_paused actor's create is cancelled (transient #account state)", + }, + { + name: "opted-out actor cancels a create", + disable: func(t *testing.T, conn *sql.DB) { + seedWorkerActor(t, conn, true, false) + _, err := store.NewFederationPrefs(conn).Upsert(context.Background(), store.FederationPref{ + DID: wActorDID, Enabled: false, Source: store.FederationPrefSourceRecord, + }) + require.NoError(t, err) + }, + kind: "Create", + wantState: store.DeliveryStateCancelled, + wantPosted: false, + explanation: "a federation opt-out cancels an outward create", + }, + { + name: "disabled actor STILL delivers a delete", + disable: func(t *testing.T, conn *sql.DB) { seedWorkerActor(t, conn, false, false) }, + kind: "Delete", + wantState: store.DeliveryStateDelivered, + wantPosted: true, + explanation: "retraction asymmetry: a Delete goes out even for a disabled actor — it is the " + + "only way an opted-out user takes down what is already federated", + }, + { + name: "disabled actor STILL delivers an undo", + disable: func(t *testing.T, conn *sql.DB) { seedWorkerActor(t, conn, false, false) }, + kind: "Undo", + wantState: store.DeliveryStateDelivered, + wantPosted: true, + explanation: "an Undo (vote retraction) is also exempt from the consent recheck", + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + conn := workerTestDB(t) + tc.disable(t, conn) + id := seedDelivery(t, conn, tc.kind, "", createPayload("x")) + + sender := &fakeSender{} + w := newWorker(t, conn, sender, nil) + worked, err := w.DeliverNext(context.Background()) + require.NoError(t, err) + assert.True(t, worked) + + assert.Equal(t, tc.wantState, getDelivery(t, conn, id).State, tc.explanation) + if tc.wantPosted { + assert.Equal(t, 1, sender.count(), "a retraction must actually be POSTed") + } else { + assert.Zero(t, sender.count(), "a cancelled delivery must POST nothing") + } + }) + } +} + +// --------------------------------------------------------------------------- +// F: vote delivery-success callbacks (decision 16) +// --------------------------------------------------------------------------- + +func TestWorker_LikeDeliverySuccessFlipsVoteState(t *testing.T) { + conn := workerTestDB(t) + seedWorkerActor(t, conn, true, false) + + likeID := "https://coves.social/ap/activity/" + repeatHex64("Like") + voteATURI := "at://" + wActorDID + "/social.coves.feed.vote/3lzvoteaaaa" + _, err := store.NewOutboundVotes(conn).Upsert(context.Background(), store.OutboundVote{ + VoteATURI: voteATURI, + ActorDID: wActorDID, + SubjectATURI: "at://" + wCommunityDID + "/social.coves.community.postv2/3lzpost", + SubjectAPID: "https://lemmy.world/post/1", + CommunityDID: wCommunityDID, + Direction: "up", + CurrentActivityID: likeID, + }) + require.NoError(t, err) + + seedDelivery(t, conn, "Like", "", []byte(fmt.Sprintf( + `{"id":%q,"type":"Like","actor":%q,"object":"https://lemmy.world/post/1"}`, likeID, wActorID))) + + w := newWorker(t, conn, &fakeSender{}, nil) + worked, err := w.DeliverNext(context.Background()) + require.NoError(t, err) + assert.True(t, worked) + + vote, err := store.NewOutboundVotes(conn).GetByATURI(context.Background(), voteATURI) + require.NoError(t, err) + assert.Equal(t, store.DeliveredStateDelivered, vote.DeliveredState, + "a Like delivery success flips outbound_votes.delivered_state to delivered (decision 16)") +} + +func TestWorker_UndoDeliverySuccessClearsVoteState(t *testing.T) { + conn := workerTestDB(t) + seedWorkerActor(t, conn, true, false) + + likeID := "https://coves.social/ap/activity/" + repeatHex64("Like") + voteATURI := "at://" + wActorDID + "/social.coves.feed.vote/3lzvotebbbb" + _, err := store.NewOutboundVotes(conn).Upsert(context.Background(), store.OutboundVote{ + VoteATURI: voteATURI, + ActorDID: wActorDID, + SubjectATURI: "at://" + wCommunityDID + "/social.coves.community.postv2/3lzpost", + SubjectAPID: "https://lemmy.world/post/1", + CommunityDID: wCommunityDID, + Direction: "up", + CurrentActivityID: likeID, + DeliveredState: store.DeliveredStateDelivered, + }) + require.NoError(t, err) + + // The Undo embeds the Like's id as its inner object.id — the worker resolves + // the vote row from that. + seedDelivery(t, conn, "Undo", "", []byte(fmt.Sprintf( + `{"id":%q,"type":"Undo","actor":%q,"object":{"type":"Like","id":%q,"actor":%q,`+ + `"object":"https://lemmy.world/post/1"}}`, + "https://coves.social/ap/activity/"+repeatHex64("Undo"), wActorID, likeID, wActorID))) + + w := newWorker(t, conn, &fakeSender{}, nil) + worked, err := w.DeliverNext(context.Background()) + require.NoError(t, err) + assert.True(t, worked) + + _, err = store.NewOutboundVotes(conn).GetByATURI(context.Background(), voteATURI) + assert.Truef(t, errors.IsNotFound(err), + "an Undo delivery success CLEARS the vote row (clear-on-Undo), got %v", err) +} + +// --------------------------------------------------------------------------- +// I: kill switches + dry-run +// --------------------------------------------------------------------------- + +func TestWorker_KillSwitchParksPending(t *testing.T) { + conn := workerTestDB(t) + seedWorkerActor(t, conn, true, false) + id := seedDelivery(t, conn, "Create", "", createPayload("x")) + + switches := &fakeSwitches{allow: false} + sender := &fakeSender{} + w := newWorker(t, conn, sender, func(o *WorkerOptions) { o.Switches = switches }) + + worked, err := w.DeliverNext(context.Background()) + require.NoError(t, err) + assert.True(t, worked) + + assert.Zero(t, sender.count(), "a kill switch POSTs nothing") + assert.Equal(t, store.DeliveryStatePending, getDelivery(t, conn, id).State, + "a killed delivery is PARKED (stays pending, resumable when the switch clears) — never "+ + "poisoned, never cancelled") + + // The switch is consulted with the full scope so global/per-host/ + // per-community/per-actor can all be expressed. + require.NotEmpty(t, switches.scopes) + scope := switches.scopes[0] + assert.Equal(t, wActorDID, scope.ActorDID, "the actor is in the scope") + assert.Equal(t, wCommunityAPID, scope.CommunityAPID, "the community is in the scope") + assert.Equal(t, "lemmy.world", scope.InboxHost, "the inbox host is in the scope") +} + +func TestWorker_DryRunPostsNothing(t *testing.T) { + conn := workerTestDB(t) + seedWorkerActor(t, conn, true, false) + id := seedDelivery(t, conn, "Create", "", createPayload("x")) + + sender := &fakeSender{} + w := newWorker(t, conn, sender, func(o *WorkerOptions) { + o.Switches = &fakeSwitches{allow: true, dryRun: true} + }) + + worked, err := w.DeliverNext(context.Background()) + require.NoError(t, err) + assert.True(t, worked) + + assert.Zero(t, sender.count(), "dry-run translates + logs but POSTs nothing") + assert.Equal(t, store.DeliveryStatePending, getDelivery(t, conn, id).State, + "dry-run does not mark delivered — the delivery stays pending") +} diff --git a/internal/personas/personas.go b/internal/personas/personas.go index 1f014b3..64a103e 100644 --- a/internal/personas/personas.go +++ b/internal/personas/personas.go @@ -238,6 +238,13 @@ func (s *Service) ActorSigner(ctx context.Context, did string) (*ap.Signer, erro return s.actorSigner(ctx, did) } +// SignerFor satisfies outbound.SignerProvider so main can inject *Service as the +// delivery worker's signer source (cycle J). It is ActorSigner under the name +// the worker's interface uses. +func (s *Service) SignerFor(ctx context.Context, did string) (*ap.Signer, error) { + return s.actorSigner(ctx, did) +} + // actorSigner unseals a minted actor's RSA key and returns a Signer whose // keyID is "{actor_id}#main-key" — the same id the actor document publishes, // so a verifier that fetches the document finds the key it needs there. diff --git a/internal/store/interfaces.go b/internal/store/interfaces.go index c3f540b..a158f77 100644 --- a/internal/store/interfaces.go +++ b/internal/store/interfaces.go @@ -373,6 +373,13 @@ type OutboundVotes interface { // A miss is an error satisfying errors.IsNotFound. GetByActorSubject(ctx context.Context, actorDID, subjectATURI string) (*OutboundVote, error) + // GetByActivityID returns the vote whose CurrentActivityID equals + // activityID — the DELIVERY callback's lookup (task 15). A Like/Dislike is + // delivered under CurrentActivityID; an Undo embeds that same id as its + // inner object, so both delivery-success callbacks resolve the vote row + // from the one activity id. A miss is an error satisfying errors.IsNotFound. + GetByActivityID(ctx context.Context, activityID string) (*OutboundVote, error) + // SetDeliveredState transitions the delivery state. An unknown state is // an error satisfying errors.IsValidation; a missing vote is an error // satisfying errors.IsNotFound. @@ -532,4 +539,12 @@ type OutboundDeliveries interface { // Get returns the delivery for an (activity, inbox) pair. A miss is an // error satisfying errors.IsNotFound. Get(ctx context.Context, activityID, targetInbox string) (*OutboundDelivery, error) + + // HasPoisonedPredecessor reports whether an earlier delivery on the same + // ordering key and target inbox (a lower seq) is poisoned — the causal + // signal task 15's worker reads to distinguish a child whose bridge-origin + // parent WILL NEVER land (parent_poisoned) from one merely waiting + // (parent_unaccepted). Per-community serialization makes a lower-seq + // delivery on the same line a causal ancestor. + HasPoisonedPredecessor(ctx context.Context, orderingKey, targetInbox string, seq int64) (bool, error) } diff --git a/internal/store/models.go b/internal/store/models.go index bdfa32d..0ef29be 100644 --- a/internal/store/models.go +++ b/internal/store/models.go @@ -246,12 +246,20 @@ type OutboundObject struct { CreatedAt time.Time UpdatedAt time.Time TombstonedAt *time.Time + // AcceptedAt is the causal-gating marker (task 15, migration 020): stamped + // on delivery success. NULL means "not yet accepted by its community", which + // gates a bridge-origin child from delivering before its parent. + AcceptedAt *time.Time } // IsTombstoned reports whether the record was deleted upstream. Tombstoned // rows are KEPT: they are what a late replay is rejected against. func (o *OutboundObject) IsTombstoned() bool { return o.TombstonedAt != nil } +// IsAccepted reports whether the object has been accepted by its community (its +// AP delivery succeeded). A bridge-origin parent gates its children until it is. +func (o *OutboundObject) IsAccepted() bool { return o.AcceptedAt != nil } + // OutboundVote is the durable outbound state for one native vote (decision // 16). A vote DELETE commit names only the vote record, so direction and the // activity id it was delivered under have to be readable back from here to diff --git a/internal/store/outbound_deliveries.go b/internal/store/outbound_deliveries.go index 3dd1044..34a070f 100644 --- a/internal/store/outbound_deliveries.go +++ b/internal/store/outbound_deliveries.go @@ -264,6 +264,20 @@ func (r *postgresOutboundDeliveries) Get(ctx context.Context, activityID, target return delivery, nil } +func (r *postgresOutboundDeliveries) HasPoisonedPredecessor(ctx context.Context, orderingKey, targetInbox string, seq int64) (bool, error) { + var exists bool + err := r.db.QueryRowContext(ctx, ` + SELECT EXISTS ( + SELECT 1 FROM outbound_deliveries + WHERE ordering_key = $1 AND target_inbox = $2 + AND state = 'poisoned' AND seq < $3)`, + orderingKey, targetInbox, seq).Scan(&exists) + if err != nil { + return false, fmt.Errorf("check poisoned predecessor on %q: %w", orderingKey, err) + } + return exists, nil +} + func scanOutboundDelivery(row rowScanner) (*OutboundDelivery, error) { var delivery OutboundDelivery var state string diff --git a/internal/store/outbound_objects.go b/internal/store/outbound_objects.go index ddd807a..6f142f2 100644 --- a/internal/store/outbound_objects.go +++ b/internal/store/outbound_objects.go @@ -21,7 +21,7 @@ func NewOutboundObjects(db *sql.DB) OutboundObjects { const outboundObjectColumns = ` at_uri, ap_object_id, last_cid, last_rev, community_did, community_ap_id, translated_snapshot, - last_activity_seq, depth, created_at, updated_at, tombstoned_at` + last_activity_seq, depth, created_at, updated_at, tombstoned_at, accepted_at` // execer is the subset of *sql.DB and *sql.Tx these repositories need, so one // statement runs either standalone or inside a caller's transaction. @@ -161,12 +161,12 @@ func (r *postgresOutboundObjects) SetAccepted(ctx context.Context, atURI string) func scanOutboundObject(row rowScanner) (*OutboundObject, error) { var object OutboundObject - var tombstonedAt sql.NullTime + var tombstonedAt, acceptedAt sql.NullTime err := row.Scan( &object.ATURI, &object.APObjectID, &object.LastCID, &object.LastRev, &object.CommunityDID, &object.CommunityAPID, &object.TranslatedSnapshot, &object.LastActivitySeq, &object.Depth, - &object.CreatedAt, &object.UpdatedAt, &tombstonedAt, + &object.CreatedAt, &object.UpdatedAt, &tombstonedAt, &acceptedAt, ) if err != nil { return nil, err @@ -174,5 +174,8 @@ func scanOutboundObject(row rowScanner) (*OutboundObject, error) { if tombstonedAt.Valid { object.TombstonedAt = &tombstonedAt.Time } + if acceptedAt.Valid { + object.AcceptedAt = &acceptedAt.Time + } return &object, nil } diff --git a/internal/store/outbound_votes.go b/internal/store/outbound_votes.go index df4ac20..f7c6185 100644 --- a/internal/store/outbound_votes.go +++ b/internal/store/outbound_votes.go @@ -109,6 +109,22 @@ func (r *postgresOutboundVotes) GetByActorSubject(ctx context.Context, actorDID, return vote, nil } +// GetByActivityID looks a vote up by its current activity id — the delivery +// callback's lookup: a Like/Dislike is delivered under CurrentActivityID and an +// Undo embeds that same id, so both success callbacks resolve the vote row from +// the one id. A miss is a NotFound. +func (r *postgresOutboundVotes) GetByActivityID(ctx context.Context, activityID string) (*OutboundVote, error) { + query := `SELECT` + outboundVoteColumns + ` FROM outbound_votes WHERE current_activity_id = $1` + vote, err := scanOutboundVote(r.db.QueryRowContext(ctx, query, activityID)) + if err != nil { + if stderrors.Is(err, sql.ErrNoRows) { + return nil, errors.NewNotFoundError("outbound_vote", activityID) + } + return nil, fmt.Errorf("get outbound_vote by activity id %q: %w", activityID, err) + } + return vote, nil +} + func (r *postgresOutboundVotes) SetDeliveredState(ctx context.Context, voteATURI string, state DeliveredState) error { // Validated in Go rather than left to the CHECK constraint: an unknown // state is a caller bug, and the caller needs it back as a validation -- 2.51.2