From 109de0384adf0f49f0a95dceb5df0dd02ae3afd6 Mon Sep 17 00:00:00 2001 From: Alan Johnson Date: Fri, 31 Jul 2026 17:55:45 -0400 Subject: [PATCH] Fix relay delivery retries and diagnostics --- ARCHITECTURE.md | 23 +- CHANGELOG.md | 19 ++ api/relay_activity_store_test.go | 7 +- api/relay_retry_policy_test.go | 32 +++ api/resolver.go | 8 +- debian/changelog | 12 ++ deliver/deliver.go | 282 +++++++++++++++++++++---- deliver/deliver_test.go | 6 +- deliver/operational_metrics_test.go | 2 +- deliver/retry_test.go | 221 +++++++++++++++++++ deliver/sender.go | 68 ++++-- deliver/sender_response_test.go | 45 ++++ docs/INTEROPERABILITY.md | 15 ++ docs/releases/v2.5.0.md | 47 +++-- internal/deliverymeta/metadata.go | 86 ++++++++ internal/deliverymeta/metadata_test.go | 63 ++++++ internal/deliverypolicy/policy.go | 23 ++ internal/deliverypolicy/policy_test.go | 32 +++ readme.md | 4 +- 19 files changed, 908 insertions(+), 87 deletions(-) create mode 100644 api/relay_retry_policy_test.go create mode 100644 deliver/retry_test.go create mode 100644 deliver/sender_response_test.go create mode 100644 internal/deliverymeta/metadata.go create mode 100644 internal/deliverymeta/metadata_test.go create mode 100644 internal/deliverypolicy/policy.go create mode 100644 internal/deliverypolicy/policy_test.go diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 5ff4926..66b2672 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -74,10 +74,20 @@ and ActivityPub object resolution also uses the relay actor identity. Authorized fetch `GET` requests sign `(request-target)`, `Host`, and `Date`; redirected fetches are re-signed for the new request target and authority. Delivery `POST` requests additionally sign `Digest` and `Content-Type`. Both paths sign the exact -`Host` authority transmitted on the wire. Each `relay-v2` fan-out attempt also -atomically records the receiver domain's last success or failure time, total -successes and failures, and consecutive failure count. Registration traffic does -not affect these delivery-health observations. +`Host` authority transmitted on the wire. Each `relay-v2` target receives one initial attempt and five delayed retries +using Machinery's Fibonacci schedule. The shared activity body remains in Redis +for fifteen minutes, which exceeds the complete retry horizon. A target reduces +the shared `remain_count` only after success or final retry exhaustion, so one +temporary receiver failure cannot delete the payload needed by its own retry or +by another receiver. Every attempt atomically records receiver delivery health; +registration traffic remains outside these observations. + +Worker logs contain bounded structured correlation fields: stable task UUID, +storage ID, public activity and object identifiers when present, origin and +receiver domains, body digest, attempt counters, elapsed time, HTTP status, +error class, next retry delay, and bounded non-success response text. Raw +activity bodies, signatures, private keys, and unbounded remote content are not +logged or exported as metric labels. The existing Fediverse `Signature`-header profile remains the compatibility baseline. RFC 9421 verification and signing are planned as an additive security @@ -208,8 +218,9 @@ configuration, inbox URLs, actor IDs, error text, and blocked lists. - Queue admission is bounded and atomic. - Remote JSON and error bodies are size-limited, including non-success actor and object fetch responses. -- Worker failures include actionable bounded response text and are retried by - the task backend. +- Worker failures include bounded structured correlation fields and response + text. Retriable attempts preserve the shared activity body; success and final + exhaustion are the only terminal target outcomes that reduce `remain_count`. - A receiver-health recording failure is logged but does not change the delivery result or retry decision. - An observability bind failure prevents the API server from accepting traffic, diff --git a/CHANGELOG.md b/CHANGELOG.md index 26508fb..e889214 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,25 @@ ## [Unreleased] +## [2.5.0-rc2] - 2026-07-31 +### Added +- Add bounded structured delivery-attempt diagnostics containing task and + activity correlation identifiers, origin and receiver domains, attempt + counters, elapsed time, HTTP status, error class, retry schedule, body digest, + and bounded non-success response text. +- Add regression tests proving a retriable receiver failure preserves the shared + activity body and a later retry can deliver it successfully. +### Changed +- Give each `relay-v2` fan-out target five delayed retries using Machinery's + Fibonacci schedule after the initial delivery attempt. +- Retain shared fan-out activity bodies for fifteen minutes, exceeding the full + retry horizon with an operational margin. +### Fixed +- Decrement a shared activity's `remain_count` only after successful delivery or + final retry exhaustion, rather than after every failed attempt. +- Prevent transient receiver timeouts from deleting the payload required by + their own delayed retries or by other receivers. + ## [2.5.0-rc1] - 2026-07-30 ### Added diff --git a/api/relay_activity_store_test.go b/api/relay_activity_store_test.go index 0a0e656..33f7309 100644 --- a/api/relay_activity_store_test.go +++ b/api/relay_activity_store_test.go @@ -4,6 +4,8 @@ import ( "context" "testing" "time" + + "github.com/thystra/Activity-Relay/internal/deliverypolicy" ) func TestStoreRelayActivityReturnsSuccessAndPersistsPayload(t *testing.T) { @@ -36,7 +38,8 @@ func TestStoreRelayActivityReturnsSuccessAndPersistsPayload(t *testing.T) { if err != nil { t.Fatalf("read relay activity TTL: %v", err) } - if ttl <= 0 || ttl > 2*time.Minute { - t.Errorf("TTL = %v; want > 0 and <= 2m", ttl) + minimumTTL := deliverypolicy.ActivityRetention - time.Minute + if ttl < minimumTTL || ttl > deliverypolicy.ActivityRetention { + t.Errorf("TTL = %v; want between %v and %v", ttl, minimumTTL, deliverypolicy.ActivityRetention) } } diff --git a/api/relay_retry_policy_test.go b/api/relay_retry_policy_test.go new file mode 100644 index 0000000..455725d --- /dev/null +++ b/api/relay_retry_policy_test.go @@ -0,0 +1,32 @@ +// File: api/relay_retry_policy_test.go +package api + +import ( + "testing" + + "github.com/thystra/Activity-Relay/internal/deliverypolicy" +) + +func TestRelayTaskUsesBoundedRetryPolicy(t *testing.T) { + signature := relayTask( + "https://receiver.example/inbox", + "activity-storage-id", + ) + + if signature.RetryCount != deliverypolicy.RetryCount { + t.Fatalf( + "RetryCount = %d; want %d", + signature.RetryCount, + deliverypolicy.RetryCount, + ) + } + if signature.RetryTimeout != deliverypolicy.InitialRetryTimeoutSeconds { + t.Fatalf( + "RetryTimeout = %d; want %d", + signature.RetryTimeout, + deliverypolicy.InitialRetryTimeoutSeconds, + ) + } +} + +// EOF: api/relay_retry_policy_test.go diff --git a/api/resolver.go b/api/resolver.go index 6b395cd..be89c5b 100644 --- a/api/resolver.go +++ b/api/resolver.go @@ -12,6 +12,7 @@ import ( "github.com/RichardKnop/machinery/v2/tasks" "github.com/google/uuid" "github.com/sirupsen/logrus" + "github.com/thystra/Activity-Relay/internal/deliverypolicy" "github.com/thystra/Activity-Relay/models" ) @@ -136,8 +137,9 @@ func enqueueRegisterActivity(inboxURL string, body []byte) { } func relayTask(inboxURL string, activityID string) *tasks.Signature { return &tasks.Signature{ - Name: "relay-v2", - RetryCount: 0, + Name: "relay-v2", + RetryCount: deliverypolicy.RetryCount, + RetryTimeout: deliverypolicy.InitialRetryTimeoutSeconds, Args: []tasks.Arg{ { Name: "inboxURL", @@ -160,7 +162,7 @@ func storeRelayActivity(activityID string, body []byte, targetCount int) error { []string{"relay:activity:" + activityID}, body, targetCount, - 2*60, + deliverypolicy.ActivityRetentionSeconds, ).Int() if err != nil { recordRedisOperationFailure("api", "activity_store") diff --git a/debian/changelog b/debian/changelog index 431f107..922b2f3 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,15 @@ +activity-relay (2.5.0~rc2-1) noble; urgency=medium + + * Retry each relay fan-out target five times after the initial attempt. + * Retain shared activity bodies for fifteen minutes across the retry horizon. + * Reduce remain_count only on successful or finally exhausted target delivery. + * Add bounded structured task, activity, receiver, timing, status, retry, and + remote-response diagnostics without logging activity bodies or key material. + * Add regression tests for retriable failure retention, successful retry, + terminal cleanup, retry horizon, metadata extraction, and response bounds. + + -- thystra Fri, 31 Jul 2026 17:23:00 -0400 + activity-relay (2.5.0~rc1-1) noble; urgency=medium * Sign remote actor and canonical-object GET requests with the relay actor diff --git a/deliver/deliver.go b/deliver/deliver.go index 52870d3..c2d461c 100644 --- a/deliver/deliver.go +++ b/deliver/deliver.go @@ -3,14 +3,19 @@ package deliver import ( "context" "errors" + "fmt" "net/http" "time" "github.com/RichardKnop/machinery/v2" "github.com/RichardKnop/machinery/v2/log" + "github.com/RichardKnop/machinery/v2/retry" + "github.com/RichardKnop/machinery/v2/tasks" "github.com/google/uuid" "github.com/redis/go-redis/v9" "github.com/sirupsen/logrus" + "github.com/thystra/Activity-Relay/internal/deliverymeta" + "github.com/thystra/Activity-Relay/internal/deliverypolicy" "github.com/thystra/Activity-Relay/internal/observability" "github.com/thystra/Activity-Relay/models" ) @@ -46,63 +51,256 @@ func recordWorkerRedisFailure(operation string) { } } -func relayActivityV2(args ...string) error { +type deliveryAttempt struct { + TaskUUID string + Attempt int + MaxAttempts int + RetriesRemaining int + NextRetrySeconds int + Terminal bool +} + +func currentDeliveryAttempt(ctx context.Context) deliveryAttempt { + signature := tasks.SignatureFromContext(ctx) + if signature == nil { + return deliveryAttempt{ + TaskUUID: "direct", + Attempt: 1, + MaxAttempts: 1, + RetriesRemaining: 0, + Terminal: true, + } + } + + retriesRemaining := signature.RetryCount + attempt := deliverypolicy.MaxAttempts - retriesRemaining + if attempt < 1 { + attempt = 1 + } + if attempt > deliverypolicy.MaxAttempts { + attempt = deliverypolicy.MaxAttempts + } + + nextRetrySeconds := 0 + if retriesRemaining > 0 { + nextRetrySeconds = retry.FibonacciNext(signature.RetryTimeout) + } + + return deliveryAttempt{ + TaskUUID: signature.UUID, + Attempt: attempt, + MaxAttempts: deliverypolicy.MaxAttempts, + RetriesRemaining: retriesRemaining, + NextRetrySeconds: nextRetrySeconds, + Terminal: retriesRemaining <= 0, + } +} + +func boundedDeliveryLogValue(value string, maximum int) string { + if len(value) <= maximum { + return value + } + return value[:maximum] +} + +func finishRelayTarget(activityID string) (int64, error) { + const finishScript = ` +if redis.call('EXISTS', KEYS[1]) == 0 then + return -1 +end +local remaining = redis.call('HINCRBY', KEYS[1], 'remain_count', -1) +if remaining < 1 then + redis.call('DEL', KEYS[1]) +end +return remaining` + + remaining, err := RedisClient.Eval( + context.Background(), + finishScript, + []string{"relay:activity:" + activityID}, + ).Int64() + if err != nil { + recordWorkerRedisFailure("activity_cleanup") + } + return remaining, err +} + +func recordReceiverAttempt(receiverDomain string, succeeded bool) { + if receiverDomain == "" { + return + } + + healthErr := models.RecordReceiverDelivery( + context.Background(), + RedisClient, + receiverDomain, + succeeded, + time.Now().UTC(), + ) + if healthErr != nil { + recordWorkerRedisFailure("receiver_health") + logrus.WithError(healthErr). + WithField("receiver", receiverDomain). + Warn("Unable to record receiver delivery health") + } +} + +func storeTransientDeliveryError(receiverDomain string, deliveryErr error) { + if receiverDomain == "" || deliveryErr == nil { + return + } + + const pushErrorLogScript = ` +local changed = redis.call('HSETNX', KEYS[1], 'last_error', ARGV[1]) +if changed == 1 then + redis.call('EXPIRE', KEYS[1], ARGV[2]) +end +return changed` + + if statisticsErr := RedisClient.Eval( + context.Background(), + pushErrorLogScript, + []string{"relay:statistics:" + receiverDomain}, + boundedDeliveryLogValue(deliveryErr.Error(), 4096), + 60, + ).Err(); statisticsErr != nil { + recordWorkerRedisFailure("statistics") + logrus.WithError(statisticsErr). + Debug("Unable to store transient delivery error") + } +} + +func relayActivityV2(ctx context.Context, args ...string) error { + if len(args) < 2 { + return fmt.Errorf("relay-v2 requires inbox URL and activity storage ID") + } + inboxURL := args[0] - activityID := args[1] - body, err := RedisClient.HGet(context.TODO(), "relay:activity:"+activityID, "body").Result() + activityStorageID := args[1] + attempt := currentDeliveryAttempt(ctx) + receiverDomain, _ := models.ReceiverDomainFromInboxURL(inboxURL) + + fields := logrus.Fields{ + "task_uuid": attempt.TaskUUID, + "activity_storage_id": activityStorageID, + "receiver": boundedDeliveryLogValue(inboxURL, 2048), + "receiver_domain": receiverDomain, + "attempt": attempt.Attempt, + "max_attempts": attempt.MaxAttempts, + "retries_remaining": attempt.RetriesRemaining, + } + + body, err := RedisClient.HGet( + context.Background(), + "relay:activity:"+activityStorageID, + "body", + ).Result() if err != nil { errorClass := "activity_expired" if !errors.Is(err, redis.Nil) { errorClass = "other" recordWorkerRedisFailure("activity_load") } + + fields["error_class"] = errorClass + fields["error"] = boundedDeliveryLogValue(err.Error(), 1024) recordDeliveryMetric("skipped", errorClass) - return errors.New("activity ttl expired") + + if !attempt.Terminal { + fields["next_retry_seconds"] = attempt.NextRetrySeconds + fields["next_retry_at"] = time.Now().UTC(). + Add(time.Duration(attempt.NextRetrySeconds) * time.Second). + Format(time.RFC3339Nano) + logrus.WithFields(fields). + Warn("Relay activity body unavailable; retry scheduled") + return fmt.Errorf("load relay activity body: %w", err) + } + + if remaining, cleanupErr := finishRelayTarget(activityStorageID); cleanupErr != nil { + logrus.WithError(cleanupErr). + WithFields(fields). + Warn("Unable to finalize expired relay target") + } else { + fields["remaining_targets"] = remaining + } + + logrus.WithFields(fields). + Error("Relay activity body unavailable after retries exhausted") + return fmt.Errorf("load relay activity body: %w", err) } - err = sendActivity(inboxURL, RelayActor.PublicKey.ID, []byte(body), GlobalConfig.ActorKey()) - if err == nil { + metadata := deliverymeta.FromBody([]byte(body)) + fields["activity_id"] = metadata.ActivityID + fields["activity_type"] = metadata.ActivityType + fields["object_id"] = metadata.ObjectID + fields["origin_actor"] = metadata.ActorID + fields["origin_domain"] = metadata.OriginDomain + fields["body_sha256"] = metadata.BodySHA256 + + logrus.WithFields(fields).Debug("Relay delivery attempt started") + started := time.Now() + + response, deliveryErr := sendActivityWithResponse( + inboxURL, + RelayActor.PublicKey.ID, + []byte(body), + GlobalConfig.ActorKey(), + ) + + elapsed := time.Since(started) + fields["elapsed_ms"] = elapsed.Milliseconds() + fields["http_status"] = response.StatusCode + fields["response_body"] = boundedDeliveryLogValue(response.Body, 4096) + fields["response_truncated"] = response.BodyTruncated + + if deliveryErr == nil { recordDeliveryMetric("success", "none") - } else { - recordDeliveryMetric("failure", observability.ClassifyDeliveryError(err)) - } - receiverDomain, domainErr := models.ReceiverDomainFromInboxURL(inboxURL) - if domainErr == nil { - healthErr := models.RecordReceiverDelivery( - context.Background(), - RedisClient, - receiverDomain, - err == nil, - time.Now().UTC(), - ) - if healthErr != nil { - recordWorkerRedisFailure("receiver_health") - logrus.WithError(healthErr).WithField("receiver", receiverDomain).Warn("Unable to record receiver delivery health") + recordReceiverAttempt(receiverDomain, true) + + remaining, cleanupErr := finishRelayTarget(activityStorageID) + if cleanupErr != nil { + logrus.WithError(cleanupErr). + WithFields(fields). + Warn("Delivery succeeded but target cleanup failed") + } else { + fields["remaining_targets"] = remaining } + + logrus.WithFields(fields).Debug("Relay delivery succeeded") + return nil } - if err != nil && domainErr == nil { - pushErrorLogScript := "local change = redis.call('HSETNX', KEYS[1], 'last_error', ARGV[1]); if change == 1 then redis.call('EXPIRE', KEYS[1], ARGV[2]) end;" - if statisticsErr := RedisClient.Eval( - context.TODO(), - pushErrorLogScript, - []string{"relay:statistics:" + receiverDomain}, - err.Error(), - 60, - ).Err(); statisticsErr != nil { - recordWorkerRedisFailure("statistics") - logrus.WithError(statisticsErr).Debug("Unable to store transient delivery error") - } + + errorClass := observability.ClassifyDeliveryError(deliveryErr) + fields["error_class"] = errorClass + fields["error"] = boundedDeliveryLogValue(deliveryErr.Error(), 4096) + + recordDeliveryMetric("failure", errorClass) + recordReceiverAttempt(receiverDomain, false) + storeTransientDeliveryError(receiverDomain, deliveryErr) + + if !attempt.Terminal { + fields["next_retry_seconds"] = attempt.NextRetrySeconds + fields["next_retry_at"] = time.Now().UTC(). + Add(time.Duration(attempt.NextRetrySeconds) * time.Second). + Format(time.RFC3339Nano) + + logrus.WithFields(fields). + Warn("Relay delivery failed; retry scheduled") + return deliveryErr } - reductionRemainCountScript := "local remain_count = redis.call('HINCRBY', KEYS[1], 'remain_count', -1); if remain_count < 1 then redis.call('DEL', KEYS[1]) end;" - if cleanupErr := RedisClient.Eval( - context.TODO(), - reductionRemainCountScript, - []string{"relay:activity:" + activityID}, - ).Err(); cleanupErr != nil { - recordWorkerRedisFailure("activity_cleanup") - logrus.WithError(cleanupErr).Debug("Unable to update relay activity remaining count") + + remaining, cleanupErr := finishRelayTarget(activityStorageID) + if cleanupErr != nil { + logrus.WithError(cleanupErr). + WithFields(fields). + Warn("Unable to finalize exhausted relay target") + } else { + fields["remaining_targets"] = remaining } - return err + + logrus.WithFields(fields). + Error("Relay delivery failed after retries exhausted") + return deliveryErr } func registerActivity(args ...string) error { diff --git a/deliver/deliver_test.go b/deliver/deliver_test.go index a082055..a3896ce 100644 --- a/deliver/deliver_test.go +++ b/deliver/deliver_test.go @@ -63,7 +63,7 @@ func TestRelayActivity(t *testing.T) { pushActivityScript := "redis.call('HSET',KEYS[1], 'body', ARGV[1], 'remain_count', ARGV[2]); redis.call('EXPIRE', KEYS[1], ARGV[3]);" RedisClient.Eval(context.TODO(), pushActivityScript, []string{"relay:activity:" + activityID.String()}, "ExampleData", remainCount, 10).Result() - err := relayActivityV2(s.URL, activityID.String()) + err := relayActivityV2(context.Background(), s.URL, activityID.String()) if err != nil { t.Fatal(err) } @@ -94,7 +94,7 @@ func TestRelayActivityNoHost(t *testing.T) { pushActivityScript := "redis.call('HSET',KEYS[1], 'body', ARGV[1], 'remain_count', ARGV[2]); redis.call('EXPIRE', KEYS[1], ARGV[3]);" RedisClient.Eval(context.TODO(), pushActivityScript, []string{"relay:activity:" + activityID.String()}, "ExampleData", remainCount, 10).Result() - err := relayActivityV2("http://nohost.example.jp", activityID.String()) + err := relayActivityV2(context.Background(), "http://nohost.example.jp", activityID.String()) if err == nil { t.Fatal("Expected error to be reported for nohost, but got nil") } @@ -127,7 +127,7 @@ func TestRelayActivityResp500(t *testing.T) { pushActivityScript := "redis.call('HSET',KEYS[1], 'body', ARGV[1], 'remain_count', ARGV[2]); redis.call('EXPIRE', KEYS[1], ARGV[3]);" RedisClient.Eval(context.TODO(), pushActivityScript, []string{"relay:activity:" + activityID.String()}, "ExampleData", remainCount, 10).Result() - err := relayActivityV2(s.URL, activityID.String()) + err := relayActivityV2(context.Background(), s.URL, activityID.String()) if err == nil { t.Fatal("Expected error to be reported for 500 response, but got nil") } diff --git a/deliver/operational_metrics_test.go b/deliver/operational_metrics_test.go index 2f891cc..a350196 100644 --- a/deliver/operational_metrics_test.go +++ b/deliver/operational_metrics_test.go @@ -32,7 +32,7 @@ func TestRelayActivityRecordsOperationalMetric(t *testing.T) { }).Err(); err != nil { t.Fatal(err) } - if err := relayActivityV2(server.URL, activityID); err == nil { + if err := relayActivityV2(context.Background(), server.URL, activityID); err == nil { t.Fatal("expected delivery failure") } ledger, err := RedisClient.HGetAll(context.Background(), observability.OperationalMetricsKey).Result() diff --git a/deliver/retry_test.go b/deliver/retry_test.go new file mode 100644 index 0000000..a7667bf --- /dev/null +++ b/deliver/retry_test.go @@ -0,0 +1,221 @@ +// File: deliver/retry_test.go +package deliver + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "testing" + + "github.com/RichardKnop/machinery/v2/tasks" + "github.com/google/uuid" + "github.com/thystra/Activity-Relay/internal/deliverypolicy" + "github.com/thystra/Activity-Relay/models" +) + +func storeRetryTestActivity(t *testing.T, body string, remainCount int) string { + t.Helper() + + activityID := uuid.NewString() + key := "relay:activity:" + activityID + err := RedisClient.HSet( + context.Background(), + key, + "body", + body, + "remain_count", + remainCount, + ).Err() + if err != nil { + t.Fatalf("store retry test activity: %v", err) + } + if err := RedisClient.Expire( + context.Background(), + key, + deliverypolicy.ActivityRetention, + ).Err(); err != nil { + t.Fatalf("expire retry test activity: %v", err) + } + t.Cleanup(func() { + RedisClient.Del(context.Background(), key) + }) + return activityID +} + +func callRelayTaskForTest( + t *testing.T, + inboxURL string, + activityID string, + retriesRemaining int, + retryTimeout int, +) error { + t.Helper() + + signature := &tasks.Signature{ + UUID: "task_" + uuid.NewString(), + Name: "relay-v2", + RetryCount: retriesRemaining, + RetryTimeout: retryTimeout, + Args: []tasks.Arg{ + {Name: "inboxURL", Type: "string", Value: inboxURL}, + {Name: "activityID", Type: "string", Value: activityID}, + }, + } + task, err := tasks.NewWithSignature(relayActivityV2, signature) + if err != nil { + t.Fatalf("create relay task: %v", err) + } + _, err = task.Call() + return err +} + +func TestRetriableFailurePreservesBodyUntilSuccessfulRetry(t *testing.T) { + RedisClient.FlushAll(context.Background()).Result() + + status := http.StatusInternalServerError + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(status) + _, _ = w.Write([]byte("temporary receiver failure")) + })) + defer server.Close() + + body := `{ + "id":"https://origin.example/activities/retry-test", + "type":"Announce", + "actor":"https://origin.example/actor", + "object":"https://origin.example/objects/retry-test" + }` + activityID := storeRetryTestActivity(t, body, 1) + key := "relay:activity:" + activityID + + err := callRelayTaskForTest( + t, + server.URL, + activityID, + deliverypolicy.RetryCount, + deliverypolicy.InitialRetryTimeoutSeconds, + ) + if err == nil { + t.Fatal("initial failure returned nil") + } + + values, err := RedisClient.HGetAll(context.Background(), key).Result() + if err != nil { + t.Fatalf("read retained activity: %v", err) + } + if values["body"] != body { + t.Fatalf("activity body was not retained after retriable failure") + } + if values["remain_count"] != "1" { + t.Fatalf( + "remain_count after retriable failure = %q; want 1", + values["remain_count"], + ) + } + + status = http.StatusAccepted + err = callRelayTaskForTest( + t, + server.URL, + activityID, + deliverypolicy.RetryCount-1, + 8, + ) + if err != nil { + t.Fatalf("successful retry returned error: %v", err) + } + + exists, err := RedisClient.Exists(context.Background(), key).Result() + if err != nil { + t.Fatalf("check successful retry cleanup: %v", err) + } + if exists != 0 { + t.Fatal("activity body remained after terminal success") + } + + domain, err := models.ReceiverDomainFromInboxURL(server.URL) + if err != nil { + t.Fatal(err) + } + health, err := models.LoadReceiverDeliveryHealth( + context.Background(), + RedisClient, + []string{domain}, + ) + if err != nil { + t.Fatal(err) + } + got := health[domain] + if got.TotalFailures != 1 || got.TotalSuccesses != 1 || got.ConsecutiveFailures != 0 { + t.Fatalf("unexpected retry health: %+v", got) + } +} + +func TestFinalFailureReducesTargetCount(t *testing.T) { + RedisClient.FlushAll(context.Background()).Result() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "terminal receiver failure", http.StatusBadGateway) + })) + defer server.Close() + + activityID := storeRetryTestActivity( + t, + `{"id":"https://origin.example/activities/final","type":"Announce"}`, + 1, + ) + + err := callRelayTaskForTest(t, server.URL, activityID, 0, 55) + if err == nil { + t.Fatal("terminal failure returned nil") + } + + exists, err := RedisClient.Exists( + context.Background(), + "relay:activity:"+activityID, + ).Result() + if err != nil { + t.Fatalf("check terminal cleanup: %v", err) + } + if exists != 0 { + t.Fatal("activity body remained after final exhausted failure") + } +} + +func TestDeliveryAttemptNumbering(t *testing.T) { + for retries, wantAttempt := range map[int]int{ + deliverypolicy.RetryCount: 1, + deliverypolicy.RetryCount - 1: 2, + 0: deliverypolicy.MaxAttempts, + } { + signature := &tasks.Signature{ + UUID: fmt.Sprintf("task-attempt-%d", retries), + RetryCount: retries, + RetryTimeout: deliverypolicy.InitialRetryTimeoutSeconds, + } + task, err := tasks.NewWithSignature( + func(ctx context.Context) error { + got := currentDeliveryAttempt(ctx) + if got.Attempt != wantAttempt { + t.Fatalf( + "retries remaining %d: attempt = %d; want %d", + retries, + got.Attempt, + wantAttempt, + ) + } + return nil + }, + signature, + ) + if err != nil { + t.Fatal(err) + } + if _, err := task.Call(); err != nil { + t.Fatal(err) + } + } +} + +// EOF: deliver/retry_test.go diff --git a/deliver/sender.go b/deliver/sender.go index 5f6e64e..5d423b1 100644 --- a/deliver/sender.go +++ b/deliver/sender.go @@ -26,22 +26,51 @@ func appendSignature(request *http.Request, body *[]byte, KeyID string, privateK return signer.SignPOST(request, *body) } +type deliveryResponse struct { + StatusCode int + Status string + Body string + BodyTruncated bool +} + func sendActivity(inboxURL string, KeyID string, body []byte, privateKey *rsa.PrivateKey) error { + _, err := sendActivityWithResponse(inboxURL, KeyID, body, privateKey) + return err +} + +func sendActivityWithResponse( + inboxURL string, + KeyID string, + body []byte, + privateKey *rsa.PrivateKey, +) (deliveryResponse, error) { + result := deliveryResponse{} + req, err := http.NewRequest(http.MethodPost, inboxURL, bytes.NewReader(body)) if err != nil { - return fmt.Errorf("create delivery request: %w", err) + return result, fmt.Errorf("create delivery request: %w", err) } + req.Header.Set("Content-Type", "application/activity+json") - req.Header.Set("User-Agent", fmt.Sprintf("%s (golang net/http; Activity-Relay %s; %s)", GlobalConfig.ServerServiceName(), version, GlobalConfig.ServerHostname().Host)) + req.Header.Set( + "User-Agent", + fmt.Sprintf( + "%s (golang net/http; Activity-Relay %s; %s)", + GlobalConfig.ServerServiceName(), + version, + GlobalConfig.ServerHostname().Host, + ), + ) req.Header.Set("Date", httpdate.Time2Str(time.Now())) + if err := appendSignature(req, &body, KeyID, privateKey); err != nil { - return fmt.Errorf("sign delivery request: %w", err) + return result, fmt.Errorf("sign delivery request: %w", err) } + resp, err := HttpClient.Do(req) if err != nil { var urlErr *url.Error errMsg := "" - if errors.As(err, &urlErr) && urlErr.Timeout() { errMsg = "Client.Timeout exceeded while awaiting headers" } else if urlErr != nil { @@ -49,41 +78,48 @@ func sendActivity(inboxURL string, KeyID string, body []byte, privateKey *rsa.Pr } else { errMsg = err.Error() } - return errors.New(inboxURL + ": " + errMsg) + return result, errors.New(inboxURL + ": " + errMsg) } defer resp.Body.Close() + + result.StatusCode = resp.StatusCode + result.Status = resp.Status + responseBody, readErr := io.ReadAll( io.LimitReader(resp.Body, maxDeliveryResponseBodyBytes+1), ) - responseTruncated := int64(len(responseBody)) > maxDeliveryResponseBodyBytes - if responseTruncated { + result.BodyTruncated = int64(len(responseBody)) > maxDeliveryResponseBodyBytes + if result.BodyTruncated { responseBody = responseBody[:maxDeliveryResponseBodyBytes] } - responseText := strings.Join(strings.Fields(string(responseBody)), " ") + result.Body = strings.Join(strings.Fields(string(responseBody)), " ") logrus.Debug(inboxURL, " ", resp.StatusCode) + if resp.StatusCode/100 != 2 { if readErr != nil { - return fmt.Errorf( + return result, fmt.Errorf( "%s: %s: unable to read response body: %w", inboxURL, resp.Status, readErr, ) } - if responseTruncated && responseText != "" { - responseText += " [truncated]" + if result.BodyTruncated && result.Body != "" { + result.Body += " [truncated]" } - if responseText != "" { - return fmt.Errorf( + if result.Body != "" { + return result, fmt.Errorf( "%s: %s: %s", inboxURL, resp.Status, - responseText, + result.Body, ) } - return errors.New(inboxURL + ": " + resp.Status) + return result, errors.New(inboxURL + ": " + resp.Status) } - return nil + return result, nil } + +// EOF: deliver/sender.go diff --git a/deliver/sender_response_test.go b/deliver/sender_response_test.go new file mode 100644 index 0000000..81e90d1 --- /dev/null +++ b/deliver/sender_response_test.go @@ -0,0 +1,45 @@ +// File: deliver/sender_response_test.go +package deliver + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestSendActivityWithResponseCapturesBoundedFailureDetails(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusFailedDependency) + _, _ = w.Write([]byte(strings.Repeat("dependency failure ", 400))) + })) + defer server.Close() + + response, err := sendActivityWithResponse( + server.URL, + RelayActor.PublicKey.ID, + []byte(`{"type":"Announce"}`), + GlobalConfig.ActorKey(), + ) + if err == nil { + t.Fatal("non-success delivery returned nil error") + } + if response.StatusCode != http.StatusFailedDependency { + t.Fatalf( + "status code = %d; want %d", + response.StatusCode, + http.StatusFailedDependency, + ) + } + if response.Body == "" { + t.Fatal("bounded response body is empty") + } + if !response.BodyTruncated { + t.Fatal("expected oversized response body to be marked truncated") + } + if len(response.Body) > int(maxDeliveryResponseBodyBytes)+len(" [truncated]") { + t.Fatalf("bounded response body length = %d", len(response.Body)) + } +} + +// EOF: deliver/sender_response_test.go diff --git a/docs/INTEROPERABILITY.md b/docs/INTEROPERABILITY.md index cfbdd6e..3e5a559 100644 --- a/docs/INTEROPERABILITY.md +++ b/docs/INTEROPERABILITY.md @@ -1,5 +1,20 @@ # ActivityPub interoperability +## Delivery retries and slow receivers + +RC2 retries each relay fan-out target five times after the initial attempt and +retains the shared activity body for the full retry horizon. Receiver-side +ActivityPub implementations may synchronously fetch actors, keys, objects, +parents, contexts, or collections before returning from `/inbox`; transient +timeouts therefore no longer become immediate terminal loss at the relay. +Retries remain at-least-once and a receiver must tolerate duplicate delivery +when it commits an activity after the relay client has timed out. + +Structured worker diagnostics record public identifiers, hashed body identity, +attempt and elapsed-time data, receiver and origin domains, HTTP status, error +classification, and bounded error responses. They intentionally exclude raw +activity bodies, signatures, and key material. + This document records interoperability behavior validated for the stable Activity-Relay `v2.4.0` release and the `v2.5.0-rc1` candidate line. The release-candidate results remain prerelease evidence; no matrix can guarantee diff --git a/docs/releases/v2.5.0.md b/docs/releases/v2.5.0.md index 8125a58..875c4c3 100644 --- a/docs/releases/v2.5.0.md +++ b/docs/releases/v2.5.0.md @@ -1,9 +1,10 @@ # Activity-Relay v2.5.0 > [!IMPORTANT] -> These notes are prepared for the `v2.5.0-rc1` release candidate. Until the -> annotated tag and release assets are published and independently validated, -> they are not evidence that the candidate is available or deployed. +> These notes are prepared for the `v2.5.0-rc2` release candidate. RC2 +> supersedes RC1 for testing after correcting the fan-out retry lifecycle and +> adding structured delivery diagnostics. Availability and deployment remain +> separate release gates. Activity-Relay v2.5.0 strengthens federation interoperability, receiver operability, observability, and asynchronous delivery durability. @@ -42,6 +43,29 @@ The migration preserves the existing ready queue, delayed retries, task JSON, result state, Redis database selection, and tested v1/v2 producer/consumer compatibility. Redis Cluster mode remains unsupported for reliable claims. +## RC2 delivery retry correction + +RC1 production diagnostics exposed a mismatch between documented retry +expectations and the fan-out task signature: `relay-v2` tasks had zero retries, +their shared activity body expired after two minutes, and every failed attempt +reduced the shared target counter. RC2 corrects that lifecycle. + +Each target now receives one initial attempt plus five Machinery retries at +Fibonacci delays of 8, 13, 21, 34, and 55 seconds. The shared activity body is +retained for fifteen minutes. A target reduces `remain_count` only after success +or final exhaustion, preserving the payload across transient receiver timeouts +and for concurrently pending receivers. + +Worker diagnostics now provide bounded structured correlation without logging +activity bodies, signatures, or key material. Fields include task UUID, storage +ID, public activity/object/actor identifiers when present, origin and receiver +domains, SHA-256 body identity, attempt and retry counters, elapsed time, HTTP +status, error class, next retry timing, and bounded non-success response text. + +The delivery model remains at-least-once. A receiver that commits an activity +after the relay client times out may see the same activity again on retry and +must deduplicate it by ActivityPub identity. + ## Status and observability `/status.json` schema version 4 includes delivery-health timestamps and counters @@ -94,13 +118,12 @@ gate until recorded separately. ## Remaining release gates -Before publication: +Before RC2 publication and production testing: -- complete the native-package install, upgrade, reinstall, rollback, identity, - persistence, and removal matrix; -- repeat applicable final Redis outage and transport checks through the exact - release-candidate binary or package; -- run final GitHub Actions, package lint, checksum, and container-manifest - validation; and -- publish and independently validate the release candidate before production - deployment. +- merge the retry-lifecycle and delivery-diagnostics changes after green CI; +- build and validate the exact RC2 Debian and multi-architecture container + artifacts; +- repeat the native-package upgrade, reinstall, rollback, identity, Redis, and + removal gates against RC2; and +- publish and independently verify RC2 before a controlled production upgrade + and high-fan-out soak test. diff --git a/internal/deliverymeta/metadata.go b/internal/deliverymeta/metadata.go new file mode 100644 index 0000000..728f18f --- /dev/null +++ b/internal/deliverymeta/metadata.go @@ -0,0 +1,86 @@ +// File: internal/deliverymeta/metadata.go +package deliverymeta + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "net/url" + "strings" +) + +const maxIdentifierBytes = 2048 + +type Metadata struct { + ActivityID string + ActivityType string + ObjectID string + ActorID string + OriginDomain string + BodySHA256 string +} + +func FromBody(body []byte) Metadata { + sum := sha256.Sum256(body) + metadata := Metadata{BodySHA256: hex.EncodeToString(sum[:])} + + var payload map[string]any + if json.Unmarshal(body, &payload) != nil { + return metadata + } + + metadata.ActivityID = bounded(identifier(payload["id"])) + metadata.ActivityType = bounded(textValue(payload["type"])) + metadata.ObjectID = bounded(identifier(payload["object"])) + metadata.ActorID = bounded(identifier(payload["actor"])) + metadata.OriginDomain = domain(metadata.ActorID) + return metadata +} + +func identifier(value any) string { + switch typed := value.(type) { + case string: + return typed + case map[string]any: + return identifier(typed["id"]) + case []any: + for _, entry := range typed { + if found := identifier(entry); found != "" { + return found + } + } + } + return "" +} + +func textValue(value any) string { + switch typed := value.(type) { + case string: + return typed + case []any: + for _, entry := range typed { + if found := textValue(entry); found != "" { + return found + } + } + } + return "" +} + +func domain(identifier string) string { + parsed, err := url.Parse(identifier) + if err != nil { + return "" + } + return strings.ToLower(strings.TrimSuffix(parsed.Hostname(), ".")) +} + +func bounded(value string) string { + value = strings.TrimSpace(value) + if len(value) <= maxIdentifierBytes { + return value + } + return value[:maxIdentifierBytes] +} + +// EOF: internal/deliverymeta/metadata.go diff --git a/internal/deliverymeta/metadata_test.go b/internal/deliverymeta/metadata_test.go new file mode 100644 index 0000000..57821fa --- /dev/null +++ b/internal/deliverymeta/metadata_test.go @@ -0,0 +1,63 @@ +// File: internal/deliverymeta/metadata_test.go +package deliverymeta + +import ( + "crypto/sha256" + "encoding/hex" + "strings" + "testing" +) + +func TestFromBodyExtractsPublicCorrelationMetadata(t *testing.T) { + body := []byte(`{ + "id":"https://origin.example/activities/1", + "type":"Announce", + "actor":"https://origin.example/users/alice", + "object":{"id":"https://origin.example/objects/2"} + }`) + + got := FromBody(body) + if got.ActivityID != "https://origin.example/activities/1" { + t.Fatalf("activity ID = %q", got.ActivityID) + } + if got.ActivityType != "Announce" { + t.Fatalf("activity type = %q", got.ActivityType) + } + if got.ActorID != "https://origin.example/users/alice" { + t.Fatalf("actor ID = %q", got.ActorID) + } + if got.OriginDomain != "origin.example" { + t.Fatalf("origin domain = %q", got.OriginDomain) + } + if got.ObjectID != "https://origin.example/objects/2" { + t.Fatalf("object ID = %q", got.ObjectID) + } + if got.BodySHA256 == "" { + t.Fatal("body SHA-256 is empty") + } +} + +func TestFromBodyKeepsDigestForMalformedJSON(t *testing.T) { + body := []byte(`not-json`) + sum := sha256.Sum256(body) + + got := FromBody(body) + if got.BodySHA256 != hex.EncodeToString(sum[:]) { + t.Fatalf("body SHA-256 = %q", got.BodySHA256) + } + if got.ActivityID != "" || got.ActorID != "" || got.ObjectID != "" { + t.Fatalf("unexpected metadata for malformed JSON: %+v", got) + } +} + +func TestFromBodyBoundsIdentifiers(t *testing.T) { + longID := "https://origin.example/" + strings.Repeat("x", maxIdentifierBytes) + body := []byte(`{"id":"` + longID + `"}`) + + got := FromBody(body) + if len(got.ActivityID) != maxIdentifierBytes { + t.Fatalf("bounded activity ID length = %d; want %d", len(got.ActivityID), maxIdentifierBytes) + } +} + +// EOF: internal/deliverymeta/metadata_test.go diff --git a/internal/deliverypolicy/policy.go b/internal/deliverypolicy/policy.go new file mode 100644 index 0000000..c29512c --- /dev/null +++ b/internal/deliverypolicy/policy.go @@ -0,0 +1,23 @@ +// File: internal/deliverypolicy/policy.go +package deliverypolicy + +import "time" + +const ( + // RetryCount is the number of delayed attempts after the initial delivery. + RetryCount = 5 + + // InitialRetryTimeoutSeconds seeds Machinery's Fibonacci retry schedule. + // Machinery advances 5 to 8 seconds for the first retry. + InitialRetryTimeoutSeconds = 5 + + // ActivityRetentionSeconds keeps the shared payload beyond the complete + // retry horizon while remaining bounded. + ActivityRetentionSeconds = 15 * 60 + + MaxAttempts = RetryCount + 1 +) + +const ActivityRetention = time.Duration(ActivityRetentionSeconds) * time.Second + +// EOF: internal/deliverypolicy/policy.go diff --git a/internal/deliverypolicy/policy_test.go b/internal/deliverypolicy/policy_test.go new file mode 100644 index 0000000..0fd0040 --- /dev/null +++ b/internal/deliverypolicy/policy_test.go @@ -0,0 +1,32 @@ +// File: internal/deliverypolicy/policy_test.go +package deliverypolicy + +import ( + "testing" + "time" + + "github.com/RichardKnop/machinery/v2/retry" +) + +func TestRetentionExceedsCompleteRetryHorizon(t *testing.T) { + timeout := InitialRetryTimeoutSeconds + var horizon time.Duration + + for range RetryCount { + timeout = retry.FibonacciNext(timeout) + horizon += time.Duration(timeout) * time.Second + } + + if horizon <= 2*time.Minute { + t.Fatalf("retry horizon = %v; want greater than the former 2m payload TTL", horizon) + } + if ActivityRetention < horizon+5*time.Minute { + t.Fatalf( + "activity retention = %v; want at least retry horizon %v plus 5m margin", + ActivityRetention, + horizon, + ) + } +} + +// EOF: internal/deliverypolicy/policy_test.go diff --git a/readme.md b/readme.md index e22416b..3cd6bfa 100644 --- a/readme.md +++ b/readme.md @@ -100,10 +100,10 @@ Set the stable release image in `.env`: ACTIVITY_RELAY_IMAGE=ghcr.io/thystra/activity-relay:2.4.0 ``` -After `v2.5.0-rc1` is published, candidate testing uses: +After `v2.5.0-rc2` is published, candidate testing uses: ```dotenv -ACTIVITY_RELAY_IMAGE=ghcr.io/thystra/activity-relay:2.5.0-rc1 +ACTIVITY_RELAY_IMAGE=ghcr.io/thystra/activity-relay:2.5.0-rc2 ``` Release candidates use their complete `-rcN` tag; prereleases do not move