A remote thread brings the replies that never reached us

The context of a remote post showed only what PrivaPub happened to hold:
replies from servers nobody here follows were never seen, and only the
ancestors were ever fetched. Now a persona opening a public remote thread
queues FetchReplies for the post and its root, at most once an hour each.

The job reads the thread's own collection first (FEP-7888 `context`, which
Mastodon 4.5+ serves with every reply at any depth; posts or, as FEP-f228
allows, the activities that made them), and otherwise the post's `replies`
(PeerTube's `comments`) and the replies' own, two levels down. At most 5
pages and 100 posts a job, signed by the instance actor, never a persona;
each post is fetched from its own origin and stored through StoreContext,
so only public and unlisted ones are kept.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LsXgEaXee4GCU1hwYgPJXw
This commit is contained in:
thepraandClaude Opus 5.5 committed 2026-10-05 04:54:38 +02:00
1 parent bbeeda7268
commit 0695c08ac3
12 files changed
+344 -5

No files matched your search

+2 -1
View File
@@ -85,7 +85,8 @@ PrivaPub/ ASP.NET Core Web API, net10.0
Moderation/ DomainBlocks (suspend / silence / reject media)
Signing/ HttpSignatures (draft-cavage sign/verify)
Inbox/ InboxReceiver (verify, queue, 202) → InboxProcessor (job) → Handlers/{Follow,Accept,Reject,
Undo,Create,Update,Delete,Like,Announce}; RemotePosts (build, fetch parents, FetchAncestors)
Undo,Create,Update,Delete,Like,Announce}; RemotePosts (build, fetch parents, FetchAncestors);
RemoteReplies (FetchReplies: a thread's `context`, else `replies` two levels down)
Outbox/ OutboxPublisher (who a post goes to), DeliveryService (queues jobs) + DeliveryJobHandler
Rendering/ ActivityPubRenderer (Mastodon @context, actors, notes, collections)
Domain/
+4
View File
@@ -213,6 +213,10 @@ Posts with a location (shown to nearby users of this server) never leave the ser
deletes, must be on its actor's origin. An embedded object from another origin is fetched from that origin.
- **Fetching.** All fetches are signed by the instance actor. They go only to public addresses, follow at most three
redirects and read at most 1 MB.
- **Threads.** A reply's missing parents are fetched, up to 10 levels. When someone here opens a public remote thread,
its replies are read from their servers, at most once an hour: the thread's `context` collection (FEP-7888) if it has
one, else its `replies` (PeerTube's `comments`) and theirs, two levels down; 5 pages and 100 posts at most. Only
public and unlisted replies are kept, each fetched from its own origin.
- **Reading our documents (SecureMode).** privapub.thepra.dev answers ActivityPub GETs only when they are signed, like
Mastodon's authorized fetch; the instance actor `/peasants/privapub` is the exception, since its key is needed first.
A browser asking for HTML is redirected to the public page instead.
@@ -111,6 +111,138 @@ namespace PrivaPub.Tests.Federation
Assert.DoesNotContain(_harness.Peer.Requests, r => r.Path == new Uri(all[2]).AbsolutePath);
}
JsonObject Served(JsonObject document)
{
_harness.Peer.Serve(new Uri(IdOf(document)).AbsolutePath, document.ToJsonString());
return document;
}
static JsonObject Reply(RemoteActor author, string text, JsonObject parent)
{
var note = PublicNote(author, $"<p>{text}</p>");
note["inReplyTo"] = IdOf(parent);
return note;
}
static async Task<Post> Held(JsonObject note)
{
var post = new Post
{
ObjectURI = IdOf(note), ActorURI = note["attributedTo"]!.GetValue<string>(), IsFederatedCopy = true, Visibility = PostVisibility.Public,
ContentHtml = note["content"]!.GetValue<string>()
};
await DB.Default.SaveAsync(post, TestContext.Current.CancellationToken);
return post;
}
// runs the FetchReplies job for the post, and those it queues for this peer's posts, until none is left; how many ran
async Task<int> RunReplies(Post first)
{
var token = TestContext.Current.CancellationToken;
var handler = new RepliesJobHandler(_harness.Db, _harness.Remote, _harness.RemotePosts, _harness.Queue);
var origin = new Uri(first.ObjectURI).GetLeftPart(UriPartial.Authority);
await _harness.Queue.EnqueueMany(new[] { RepliesJobHandler.For(first, 1) }, token);
var ran = 0;
while (true)
{
var mine = new List<Job>();
foreach (var job in await DB.Default.Find<Job>().Match(j => j.Kind == JobKind.FetchReplies && j.State == JobState.Pending).ExecuteAsync(token))
{
var post = await DB.Default.Find<Post>().OneAsync(JsonSerializer.Deserialize<RepliesPayload>(job.Payload)!.PostId, token);
if (post?.ObjectURI.StartsWith(origin) == true)
mine.Add(job);
}
if (mine.Count == 0)
return ran;
foreach (var job in mine)
{
Assert.Equal(JobResult.Done, (await handler.Handle(job, token)).Result);
await DB.Default.Update<Job>().MatchID(job.ID).Modify(j => j.State, JobState.Done).ExecuteAsync(token);
ran++;
}
}
}
static async Task<Dictionary<string, Post>> Stored(params JsonObject[] notes)
{
var uris = notes.Select(IdOf).ToList();
return (await DB.Default.Find<Post>().Match(p => uris.Contains(p.ObjectURI)).ExecuteAsync(TestContext.Current.CancellationToken))
.ToDictionary(p => p.ObjectURI);
}
[Fact]
public async Task A_threads_replies_are_read_from_their_servers_two_levels_down_and_only_public_ones_kept()
{
var poster = new RemoteActor(_harness.Peer, "poster");
var answerer = new RemoteActor(_harness.Peer, "answerer");
var root = PublicNote(poster, "<p>the root</p>");
// what a post carries is as old as the post: the collection is read again by its id
root["replies"] = new JsonObject
{
["id"] = IdOf(root) + "/replies", ["type"] = "Collection", ["first"] = new JsonObject { ["type"] = "CollectionPage", ["items"] = new JsonArray() }
};
Served(root);
var first = Reply(answerer, "first answer", root);
var second = Served(Reply(poster, "second answer", root));
var hidden = Reply(answerer, "for followers", root);
hidden["to"] = new JsonArray(answerer.Id + "/followers");
hidden["cc"] = new JsonArray();
Served(hidden);
var nested = Reply(poster, "nested", first);
var deeper = Served(Reply(answerer, "too deep", nested));
first["replies"] = IdOf(first) + "/replies";
Served(first);
nested["replies"] = IdOf(nested) + "/replies";
Served(nested);
Served(new JsonObject { ["id"] = IdOf(root) + "/replies", ["type"] = "Collection", ["first"] = IdOf(root) + "/replies/1" });
Served(new JsonObject
{
["id"] = IdOf(root) + "/replies/1", ["type"] = "CollectionPage", ["next"] = IdOf(root) + "/replies/2",
["items"] = new JsonArray(first.DeepClone(), IdOf(second))
});
Served(new JsonObject { ["id"] = IdOf(root) + "/replies/2", ["type"] = "CollectionPage", ["items"] = new JsonArray(IdOf(hidden)) });
Served(new JsonObject { ["id"] = IdOf(first) + "/replies", ["type"] = "OrderedCollection", ["orderedItems"] = new JsonArray(IdOf(nested)) });
Served(new JsonObject { ["id"] = IdOf(nested) + "/replies", ["type"] = "OrderedCollection", ["orderedItems"] = new JsonArray(IdOf(deeper)) });
var held = await Held(root);
Assert.Equal(3, await RunReplies(held));
var stored = await Stored(first, second, hidden, nested, deeper);
Assert.Equal(new[] { IdOf(first), IdOf(second), IdOf(nested) }.Order(), stored.Keys.Order());
Assert.Equal(held.ID, stored[IdOf(first)].AnsweringToPostId);
Assert.Equal(held.ID, stored[IdOf(second)].AnsweringToPostId);
Assert.Equal(stored[IdOf(first)].ID, stored[IdOf(nested)].AnsweringToPostId);
Assert.DoesNotContain(_harness.Peer.Requests, r => r.Path == new Uri(IdOf(nested) + "/replies").AbsolutePath);
Assert.All(_harness.Peer.Requests.Where(r => r.Path.EndsWith("/replies")), r => Assert.False(string.IsNullOrEmpty(r.Signature)));
}
[Fact]
public async Task A_threads_context_collection_brings_every_reply_at_once()
{
var poster = new RemoteActor(_harness.Peer, "poster");
var answerer = new RemoteActor(_harness.Peer, "answerer");
var root = PublicNote(poster, "<p>a thread</p>");
var context = NewId(poster, "contexts");
root["context"] = context;
root["replies"] = NewId(poster, "never");
Served(root);
var answer = Served(Reply(answerer, "an answer", root));
var deep = Served(Reply(poster, "an answer to it", answer));
// FEP-f228 lets a context list the activities that made the thread
Served(new JsonObject
{
["id"] = context, ["type"] = "OrderedCollection",
["first"] = new JsonObject { ["type"] = "OrderedCollectionPage", ["orderedItems"] = new JsonArray(IdOf(root), Create(answerer, answer), IdOf(deep)) }
});
var held = await Held(root);
Assert.Equal(1, await RunReplies(held));
var stored = await Stored(answer, deep);
Assert.Equal(held.ID, stored[IdOf(answer)].AnsweringToPostId);
Assert.Equal(stored[IdOf(answer)].ID, stored[IdOf(deep)].AnsweringToPostId);
Assert.DoesNotContain(_harness.Peer.Requests, r => r.Path == new Uri(root["replies"]!.GetValue<string>()).AbsolutePath);
}
async Task<(LocalActor Alice, RemoteActor Mallory, Post Poll)> LocalPoll()
{
var token = TestContext.Current.CancellationToken;
@@ -1,5 +1,6 @@
using MongoDB.Entities;
using PrivaPub.Models.Jobs;
using PrivaPub.Models.User;
using PrivaPub.Tests.Support;
using PrivaPub.Tests.Support.Host;
@@ -133,6 +134,36 @@ namespace PrivaPub.Tests.Http
Assert.Equal(1, (await alice.Client.Get($"/api/v1/statuses/{parent.ID}")).Ok().Body.Number("replies_count"));
}
// a remote thread opened by a persona has its replies read from its server (here its FEP-7888 context), at most hourly
[Fact]
public async Task Opening_a_remote_thread_reads_the_replies_that_never_reached_us()
{
var token = TestContext.Current.CancellationToken;
var alice = await _host.Mastodon("alice");
var bob = new RemoteActor(_peer, "bob");
var carol = new RemoteActor(_peer, "carol");
var context = $"{bob.Origin()}/contexts/{Guid.NewGuid():N}";
var post = await _host.PublicPostFrom(bob, alice, shape: note => note["context"] = context);
var answer = carol.Create("<p>said elsewhere</p>", new[] { MastodonHelpers.Public }, new[] { carol.Id + "/followers" },
note => note["inReplyTo"] = post.ObjectURI)["object"]!;
var answerId = answer["id"]!.GetValue<string>();
_peer.Serve(new Uri(answerId).AbsolutePath, answer.ToJsonString());
_peer.Serve(new Uri(context).AbsolutePath, new JsonObject
{
["id"] = context, ["type"] = "OrderedCollection", ["orderedItems"] = new JsonArray(post.ObjectURI, answerId)
}.ToJsonString());
async Task<int> Queued() => (await DB.Default.Find<Job>().Match(j => j.Kind == JobKind.FetchReplies && j.Payload.Contains(post.ID)).ExecuteAsync(token)).Count;
(await _host.Client().Get($"/api/v1/statuses/{post.ID}/context")).Ok();
Assert.Equal(0, await Queued());
Assert.Empty((await alice.Client.Get($"/api/v1/statuses/{post.ID}/context")).Ok().Body["descendants"]!.AsArray());
Assert.Equal(1, await _host.Run(j => j.Kind == JobKind.FetchReplies && j.Payload.Contains(post.ID), token));
var descendants = (await alice.Client.Get($"/api/v1/statuses/{post.ID}/context")).Ok().Body["descendants"]!.AsArray();
Assert.Equal(answerId, Assert.Single(descendants).Text("uri"));
Assert.Equal(1, await Queued());
}
// Akkoma marks a post with sensitive media "sensitive" and leaves its summary empty: the media is hidden, the words are
// not, and no warning is made up for them (found by decePub's e2e tests on the town)
[Fact]
@@ -203,6 +203,11 @@ namespace PrivaPub.Api.Mastodon.Controllers
frontier.Add(child.ID);
}
}
// a persona opening a public remote thread has its replies read from their servers, for the next look (once an hour)
if (MyId != default)
await _jobs.EnqueueMany(new[] { post, ancestors.FirstOrDefault() }
.Where(p => p is { IsFederatedCopy: true, Visibility: PostVisibility.Public or PostVisibility.Unlisted }).DistinctBy(p => p.ID)
.Select(p => Federation.Inbox.RepliesJobHandler.For(p, 1)), token);
return Json(new Context
{
Ancestors = await _mapper.Statuses(ancestors, MyId, token),
+158
View File
@@ -0,0 +1,158 @@
using System.Text.Json;
using System.Text.Json.Nodes;
using MongoDB.Entities;
using PrivaPub.Federation.Actors;
using PrivaPub.Infrastructure.Http;
using PrivaPub.Infrastructure.Jobs;
using PrivaPub.Models.Federation;
using PrivaPub.Models.Jobs;
using PrivaPub.Models.Post;
using PrivaPub.StaticServices;
using static PrivaPub.Federation.Objects.ActivityJson;
using PostEntity = PrivaPub.Models.Post.Post;
namespace PrivaPub.Federation.Inbox
{
public sealed record RepliesPayload(string PostId, int Depth);
// What was said under a remote post on other servers, read when a persona opens its thread, so the thread shows more
// than what happened to reach us. The thread's own collection comes first (FEP-7888's `context`, which Mastodon 4.5+
// serves with every reply at any depth); otherwise the post's `replies` (PeerTube's `comments`), and the replies'
// own, two levels down. Read with the instance actor's signature, never a persona's, a few pages and at most
// MaxItems posts a job; each one is fetched from its own server and stored like any post fetched for a thread
// (RemotePosts.StoreContext: public and unlisted only, its author checked against its origin).
public class RepliesJobHandler : IJobHandler
{
public const int MaxPages = 5;
public const int MaxItems = 100;
public const int MaxDepth = 2;
public const int MaxBranches = 20;
static readonly HashSet<string> CollectionTypes = new(StringComparer.Ordinal)
{
"Collection", "OrderedCollection", "CollectionPage", "OrderedCollectionPage"
};
readonly DbEntities _dbEntities;
readonly IRemoteActorService _remoteActors;
readonly IRemotePosts _remotePosts;
readonly IJobQueue _queue;
public RepliesJobHandler(DbEntities dbEntities, IRemoteActorService remoteActors, IRemotePosts remotePosts, IJobQueue queue)
{
_dbEntities = dbEntities;
_remoteActors = remoteActors;
_remotePosts = remotePosts;
_queue = queue;
}
public JobKind Kind => JobKind.FetchReplies;
public int Concurrency => 2;
public int MaxAttempts => 2;
public int PerHostLimit => 1;
// a thread is read again at most once an hour
public static Job For(PostEntity post, int depth) => new()
{
Kind = JobKind.FetchReplies,
Payload = JsonSerializer.Serialize(new RepliesPayload(post.ID, depth)),
Host = new Uri(post.ObjectURI).Host,
DedupeKey = $"replies|{post.ID}|{DateTime.UtcNow:yyyyMMddHH}"
};
public async Task<JobOutcome> Handle(Job job, CancellationToken token)
{
var payload = JsonSerializer.Deserialize<RepliesPayload>(job.Payload);
var post = await _dbEntities.Posts.MatchID(payload.PostId).ExecuteFirstAsync(token);
if (post is not { IsFederatedCopy: true, Visibility: PostVisibility.Public or PostVisibility.Unlisted } || post.DeletedAt.HasValue
|| string.IsNullOrEmpty(post.ObjectURI))
return JobOutcome.Done;
if (await Document(post, token) is not { } document)
return JobOutcome.Done;
if (await Collection(document["context"], token) is { } thread)
{
await StoreAll(thread, token);
return JobOutcome.Done;
}
if (await Collection(document["replies"] ?? document["comments"], token) is not { } replies)
return JobOutcome.Done;
var answers = await StoreAll(replies, token);
if (payload.Depth < MaxDepth)
await _queue.EnqueueMany(answers.Where(a => a.IsFederatedCopy && a.AnsweringToPostId == post.ID).Take(MaxBranches)
.Select(a => For(a, payload.Depth + 1)), token);
return JobOutcome.Done;
}
// the post as it reached us, or as its server serves it now when we kept no copy
async Task<JsonObject> Document(PostEntity post, CancellationToken token)
{
var record = await DB.Default.Find<ObjectRecord>().Match(r => r.ObjectURI == post.ObjectURI && r.Raw != null && !r.RawTruncated)
.Project(r => new ObjectRecord { Raw = r.Raw }).ExecuteFirstAsync(token);
if (record != default)
return JsonNode.Parse(record.Raw) as JsonObject;
using var scope = HttpScope.For("replies");
using var fetched = await _remoteActors.FetchObject(post.ObjectURI, token);
return fetched == default ? default : JsonNode.Parse(fetched.Root.GetRawText()) as JsonObject;
}
// a collection as its server serves it now: what a post carries is read again by its id, since an embedded first
// page is as old as the post; a value that is no web address (Pleroma's `context` is a tag: URI), or no collection,
// gives nothing
async Task<JsonObject> Collection(JsonNode node, CancellationToken token)
{
var uri = Id(node);
if (uri != default && Uri.TryCreate(uri, UriKind.Absolute, out var parsed) && (parsed.Scheme == Uri.UriSchemeHttps || parsed.Scheme == Uri.UriSchemeHttp))
{
using var scope = HttpScope.For("replies");
using var fetched = await _remoteActors.FetchObject(uri, token);
node = fetched == default ? default : JsonNode.Parse(fetched.Root.GetRawText());
}
return node is JsonObject collection && Value(collection, "type") is { } type && CollectionTypes.Contains(type) ? collection : default;
}
// every post the collection lists, stored when we lack it: its pages in order, at most MaxPages and MaxItems
async Task<List<PostEntity>> StoreAll(JsonObject collection, CancellationToken token)
{
var posts = new List<PostEntity>();
var seen = 0;
var page = collection["orderedItems"] is JsonArray || collection["items"] is JsonArray ? collection : await Page(collection["first"], token);
for (var pages = 0; page != default && pages < MaxPages && seen < MaxItems; pages++)
{
foreach (var item in Items(page))
{
if (++seen > MaxItems)
break;
// a context may list the activities that made the thread (FEP-f228) rather than its posts
var id = Value(item, "type") is "Create" or "Update" ? Id(item["object"]) : Id(item);
if (id == default)
continue;
var post = await _dbEntities.Posts.Match(p => p.ObjectURI == id && !p.DeletedAt.HasValue).ExecuteFirstAsync(token)
?? await _remotePosts.StoreContext(id, 0, token);
if (post != default)
posts.Add(post);
}
page = page["next"] is { } next ? await Page(next, token) : default;
}
return posts;
}
async Task<JsonObject> Page(JsonNode node, CancellationToken token)
{
if (node is JsonValue && Id(node) is { } uri)
{
using var scope = HttpScope.For("replies");
using var fetched = await _remoteActors.FetchObject(uri, token);
node = fetched == default ? default : JsonNode.Parse(fetched.Root.GetRawText());
}
return node as JsonObject;
}
static IEnumerable<JsonNode> Items(JsonNode page) =>
(page["orderedItems"] as JsonArray ?? page["items"] as JsonArray ?? new JsonArray()).Where(i => i != default);
}
}
@@ -44,7 +44,7 @@ namespace PrivaPub.Infrastructure.Statistics
SingleReader = false
});
static readonly TimeSpan TouchInterval = TimeSpan.FromHours(1);
static readonly HashSet<string> TouchingPurposes = new(StringComparer.Ordinal) { "actor", "key", "object", "webfinger", "context" };
static readonly HashSet<string> TouchingPurposes = new(StringComparer.Ordinal) { "actor", "key", "object", "webfinger", "context", "replies" };
readonly InteractionSalts _salts;
readonly IOptionsMonitor<StatisticsOptions> _options;
@@ -109,6 +109,7 @@ namespace PrivaPub.Middleware
.AddSingleton<InstanceDescriber>()
.AddSingleton<IJobHandler>(services => services.GetRequiredService<InstanceDescriber>())
.AddSingleton<IJobHandler, AncestorsJobHandler>()
.AddSingleton<IJobHandler, Federation.Inbox.RepliesJobHandler>()
.AddSingleton<IJobQueue, JobQueue>()
.AddSingleton<IHostCircuitBreaker, HostCircuitBreaker>()
.AddSingleton<IJobHandler, DeliveryJobHandler>()
+2 -1
View File
@@ -32,7 +32,8 @@ namespace PrivaPub.Models.Jobs
CrawlInstance,
SendRecovery,
CountAccount,
PublishScheduled
PublishScheduled,
FetchReplies
}
public enum JobState
@@ -11,7 +11,7 @@ namespace PrivaPub.Models.Statistics
public string Channel { get; set; }//recv | in | out | http | preview | crawl
public string Activity { get; set; }//Create, Follow, ...; "other" when it is no plain type name
public string Object { get; set; }//Note, Article, Person, ...
public string Purpose { get; set; }//http and crawl: actor, key, object, context, webfinger, nodeinfo, ...
public string Purpose { get; set; }//http and crawl: actor, key, object, context, replies, webfinger, nodeinfo, ...
public string Trigger { get; set; }//http: what caused the request (inbox, deliver, describe, request, ...)
public string Outcome { get; set; }//queued, refused, accepted, dropped, ok, failed, deferred, retry, dead
public string Reason { get; set; }
+1
View File
@@ -147,6 +147,7 @@ Priorities, used throughout:
| Inbound `Block`: stop delivering, hide (**done** 2026-10-04) | P2 | `relationship.blocked_by` |
| `Add`/`Remove` featured (pins, tags) | P2 | `GET /accounts/:id/statuses?pinned=true` |
| Remote like and boost totals | P2 | counts |
| Read the thread's `context` collection to complete a thread (**done** 2026-10-05, `FetchReplies`) | P2 | `/statuses/:id/context` |
| Publish `context` and a paged `replies` | P2 | — |
| `FeatureRequest`: send `Reject` (or implement FEP-7aa9) | P3 | — |
+6 -1
View File
@@ -606,7 +606,12 @@ it, raw where it doesn't.
#### P7 Threads, communities and the social graph
- **Thread backfill:**
- read in order: `contextHistory`, then `context` (paged, with ETag), then `replies`;
- read in order: `contextHistory`, then `context` (paged, with ETag), then `replies`. **Done (2026-10-05), but for
`contextHistory` and ETags:** a persona opening a public remote thread queues `FetchReplies` for the post and its
root, at most hourly each. The job reads the thread's FEP-7888 `context` collection (posts or, as FEP-f228 allows,
the activities that made them), else the post's `replies` (PeerTube's `comments`) and theirs, two levels down. It
reads at most 5 pages and 100 posts per job, signed by the instance actor, and stores what it lacks through
`StoreContext` (public and unlisted only, the author checked against its origin);
- group by the root post;
- publish our own `context` and a paged `replies`.
- **Lemmy, PieFed and Mbin:**