feat: peer clustering over the AoI grid with a NATS feed

[mikhail-dcl]

Pulse becomes the sole author of peer clusters, derived from the same grids and SnapshotBoard that drive area-of-interest, and feeds NATS with what archipelago-core publishes today. No client protocol change, no worker involvement, nothing added to the per-tick or per-packet path.

Design and rationale: docs/clustering-on-aoi.md. Plan of record: Notion "Archipelago ⇒ Pulse migration plan", iteration 1.

What ships

• ClusterTracker — BackgroundService on its own thread, one pass per second: weighted union-find with path halving over occupied grid cells, 8-neighbour adjacency, run one realm at a time. Sticky IDs plus a dwell debounce for stability. Buffers are fields cleared between passes; no LINQ.
• ClusterBoard — the immutable ClusterPass swapped in with one Volatile.Write; readers lock-free.
• NatsPublisher — sole connection owner, publish-only, fail-soft, coalescing outbox.
• Config — Clusters and Nats. Clusters:Enabled ships true, Nats:Url ships empty, so the default is shadow mode: the tracker runs and reports metrics but publishes nothing until a URL is injected. Clearing the URL is the rollback.

"Island" now means archipelago's concept and the wire contracts carrying it only — engine.islands, IslandStatusMessage, IslandChangedMessage, the /islands stats paths. Clusters are uncapped, realm-partitioned, and carry no transport details.

Realm isolation

Realm used to be a read-time filter: a per-candidate string.Equals in both AoI implementations, plus per-pass realm interning and a per-cell member partition in the tracker. It is now structural — RealmSpatialGrids holds one SpatialGrid per realm and routes each peer into exactly one, so an observer resolves its own realm's grid and every candidate it finds is already same-realm. Seven realm comparisons leave the read paths and NodeKey drops its realm.

• Per-peer bookkeeping stays global (one array pair, not one per grid): realm names are client-supplied and only length-validated, so a per-grid array indexed by PeerIndex would let one peer mint unbounded full-size arrays by teleporting to fresh names. A grid is dropped with its last occupant, bounding live grids by connected peers.
• One write lock shared by all realms — the same contention profile as the previous global grid. Set adds to the new cell before vacating the old, so the peer is never momentarily in neither, and a solo peer changing cells cannot evict the grid it is moving within.
• A cross-realm teleport vacates the old grid before publishing the snapshot. Publishing first leaves a peer whose snapshot names the new realm sitting in the old realm's grid, which an observer there reads as a cross-realm subject and holds until the stale-view sweep.
• SnapshotBoard.Publish now returns the ledger-resolved snapshot (public API change), so the publisher picks a grid without a second seqlock read. SpatialAreaOfInterest and its options are deleted — never DI-registered, no tests, config section bound to nothing.

NATS output

Subject	Payload	Cadence
peer.{addr}.cluster_change	decentraland.pulse.PeerClusterChange { cluster_id, realm }	per published assignment change
engine.islands	kernel.comms.v3.IslandStatusMessage	per pass
engine.discovery	kernel.comms.v3.ServiceDiscoveryMessage	timer, default 10 s

Gatekeeper subscribes to cluster_change, mints the LiveKit conn-string and re-emits the existing island_changed, so WS Connector and clients are untouched. PeerClusterChange is deliberately not IslandChangedMessage: that carries conn_str (a signed LiveKit JWT) and its producer runs a ban check, both of which stay with gatekeeper.

The broker URL is read from either Nats__Url or the flat NATS_URL archipelago's services use, so one injected secret serves both; Nats__Url wins.

The outbox is not one queue. engine.islands is a whole-world snapshot, so a newer one fully replaces an undelivered one — one latest-wins slot. cluster_change supersedes only per peer, so those are held one entry per peer. A shared FIFO with oldest-first eviction, the original design, could discard peer A's assignment to admit peer B's, leaving A on a stale cluster until a reassignment that may never come if A stops moving. Loss now needs more than Nats:ChannelCapacity distinct peers pending at once, and is counted separately (dropped) from benign superseding (superseded).

Topology is emitted before the events referencing it, but best-effort only: gatekeeper and stats are separate subscribers, so consumers must tolerate an unknown cluster id. Per peer, its own events are ordered.

Reconnection. Client defaults suit a fail-soft feed (unlimited retries, 2–5 s jittered backoff), plus two additions: IgnoreAuthErrorAbort = true, since by default the client stops reconnecting permanently after the same auth error twice — turning a rotated credential into a feed only a restart recovers; and a supervision loop rebuilding after 5 s, since broker loss is handled inside the client and the pipeline exiting therefore means it faulted.

Benchmarks

src/DCLPulseBenchmarks, -c Release. BenchmarkDotNet 0.15.8, .NET 10.0.10, Ryzen 9 9955HX, X64 RyuJIT x86-64-v4.

Clustering pass — ClusterTrackerBenchmarks. Cold is Pass + churn − Churn only, a difference of means, so its error is both rows' summed. Quote cold: Pass repeats over an unchanging grid and stays cached, which a second of production traffic does not.

Scenario	Peers	Warm	Cold	Alloc/pass	Clusters
Sporadic	100	7.10 µs	7.55 µs	7.6 KB	sparse + singletons
Chained	1 000	27.78 µs	30.82 µs	56 KB	1 (transitive chain)
DenseAndSparse	1 000	31.41 µs	34.55 µs	57 KB	9
CeilingUniform	4 095	320.9 µs	394.8 µs	230 KB	2

At the 4 095-peer ceiling that is ~0.04 % of one core at 1 Hz. Allocation is dominated by the immutable ClusterPass, which readers hold by reference and so cannot be pooled.

CeilingUniform documents a limit, not a win: cell-adjacency clustering is site percolation, and 4 095 peers over Genesis City at 100 u cells occupy ≈ 0.83 of cells against a Moore threshold of ≈ 0.407, so the partition collapses to 2 clusters with 4 091 peers in one. Downstream room sharding becomes load-bearing rather than an overflow path — see §3.2.

Feed encoding — NatsEncodeBenchmarks. Justifies publishing straight into the client's writer rather than ToByteArray(), which allocated, walked the message twice and memcpy'd:

Payload	ToByteArray + copy	Straight into the writer
cluster_change (13 B)	51.00 ns / 104 B	29.83 ns / 0 B
topology (44 KB)	28.18 µs / 44.3 KB	19.51 µs / 0 B

An intermediate pooled-bytes shape removed the allocation but not the double walk (36.12 ns / 28.10 µs) — only a benchmark separates it from zero-copy.

Metrics

Nine series on the existing /metrics, types as emitted by PrometheusFormatter.

Metric	Type	Notes
dcl_pulse_clusters	gauge	Zero with peers connected ⇒ no peer has a realm, or clustering is off
dcl_pulse_cluster_passes_total	counter	~1/s expected; below that the tracker is stalling
dcl_pulse_cluster_pass_duration_us_total	counter	Sum half of a sum/count pair — divide by passes for the mean
dcl_pulse_cluster_reassignments_total	counter	Post-debounce. High with a stable cluster count ⇒ flapping; raise DwellPasses
dcl_pulse_nats_published_total	counter	Delivered to the broker
dcl_pulse_nats_dropped_total	counter	Genuinely lost — the actionable one; a peer may be on a stale cluster
dcl_pulse_nats_superseded_total	counter	Replaced pre-delivery. Expected under load; freshness degrades, nothing lost
dcl_pulse_nats_reconnects_total	counter	Steady growth ⇒ flapping broker or network path
dcl_pulse_nats_connected	gauge	0 is correct in shadow mode, the shipped default

Grafana. Scrape /metrics on port 5000; supply authorization.credentials when MetricsBearerToken is set. Add a "Clusters" row to pulse-server-dashboard.json: dcl_pulse_clusters; mean pass duration as rate(…pass_duration_us_total[5m]) / rate(…passes_total[5m]); rate() of passes, reassignments and published; dropped and superseded on one panel, since the pair is only meaningful read together; dcl_pulse_nats_connected as a stat with 0 → Down. Use rate() on every _total, not irate(), at a 1 Hz production rate.

Alerts worth having: rate(dropped[5m]) > 0 for 5m; passes < 0.5/s for 5m; mean pass duration > 500 000 µs for 10m. A connected == 0 alert must be scoped to deployments where Nats:Url is set — Prometheus cannot see whether a URL is configured, so ungated it fires everywhere by design.

Decommissioned with archipelago-core (gone after cutover)

Metric	Replacement
dcl_archipelago_islands_count	dcl_pulse_clusters (gauge, clusters derived by the last pass)
dcl_archipelago_peers_count	dcl_pulse_cluster_peers (gauge; counts clustered peers — excludes peers with no realm yet or no wallet in IdentityBoard, so it can read lower than connected peers)

Dashboards using dcl_archipelago_islands_count and/or dcl_archipelago_peers_count:

1. Comms — Folder: Core Team: Catalysts
   → Open
   • Peers Over Time (timeseries) — sum(dcl_archipelago_peers_count{instance="$instance"}) by (instance)
   • Islands Over Time (timeseries) — sum(dcl_archipelago_islands_count{instance="$instance"}) by (instance)

2. Main realm comms — Folder: Core Team: Comms Service
   → Open
   • sum(dcl_archipelago_peers_count{catalyst=""})
   • sum(dcl_archipelago_islands_count{catalyst=""})

3. Online users — Folder: Core Team: Services
   → Open
   • Online users (timeseries) — sum(dcl_archipelago_peers_count{catalyst=~"peer-.*"}) by (catalyst)
   • Online users (delta) (timeseries) — sum(delta(dcl_archipelago_peers_count[$__interval:])) by (catalyst)
   • Online users (timeseries) — sum(dcl_archipelago_peers_count)

4. Performance — Folder: World team
   → Open
   • Connected users all clients (timeseries) — sum(max_over_time(dcl_archipelago_peers_count[$__interval:]))
   • Connected users E@ (timeseries) — sum(max_over_time(dcl_archipelago_peers_count{service="archipelago-ea-core"}[$__interval:]))

Verification

• dotnet build clean; 527 tests pass, 0 failures.
• All three Docker images build — required by the NATS.Client.Core package addition.
• Live broker: engine.islands at 1/s, engine.discovery on its timer, connected 1, dropped 0.
• Live reconnection: stopping and restarting the broker gave connected 1 → 0 → 1, reconnects 1, publishing resumed, dropped 0 — the outbox retained each peer's latest assignment across the outage.
• Live shadow mode: with no URL the tracker runs, the publisher exits with NATS feed disabled, connected 0, no errors.

peer.{addr}.cluster_change is unit-tested but not live-verified — it needs an authenticated peer.

Three defects came from tests rather than review: sticky-ID inheritance read the published assignment instead of the previous pass's computed one, starving the debounce so a mid-debounce fragment was minted a fresh ID every pass and could never be reassigned; the shared-outbox eviction above; and the teleport ordering under Realm isolation, which the deleted per-candidate realm filter had been masking.

Dependency

Requires protocol#453. ServiceStatus/ServiceDiscoveryMessage were never merged to @dcl/protocol main — archipelago reaches them only via a pinned CDN branch build whose source commit no longer exists. Proto regeneration here fails without it.

pulse_clusters.proto (carrying PeerClusterChange) still needs its own protocol PR; gatekeeper also needs the generated TS types to subscribe.

Deployment order

1. Deploy definitions
2. Deploy archipelago-workers (new ws-connector + stats image). Inert with respect to the cutover — WS Connector behavior is unchanged and stats still consumes heartbeats — but it lands the protocol re-pin, the subject-address normalization fix, and the contract tests before anything starts publishing the new feeds.
3. Deploy comms-gatekeeper (with the definitions PR merged, CLUSTER_SUBSCRIBER_ENABLED=true + NATS_URL arrive at deploy). It connects and idles — nobody publishes cluster_change yet. This must precede Pulse because NATS is at-most-once: events published with no subscriber simply vanish, and clients would silently get no conn-strings. Verify: NATS connected, zero received, no errors.
4. Stop archipelago-core (infra scales the archipelago-ea-core task to zero — the repo's deploy jobs are gone, but the runbook notes the running task keeps its last image until infra retires it). This must happen before Pulse's feed goes live, or you get dueling publishers: core and gatekeeper both emitting engine.peer.{addr}.island_changed and two conflicting engine.islands topologies — clients flapping between I{n} and island-C{n} rooms. The cost of this ordering is a short gap where new sessions get no island assignment (existing sessions keep their rooms — tokens are already delivered), so plan the window; don't do it mid-event.
5. Deploy Pulse (Clusters__Enabled=true + NATS_URL). This is the cutover moment — the feed starts publishing and the chain closes: cluster_change → gatekeeper mints → island_changed → WS Connector → client. Verify immediately: dcl_pulse_nats_connected 1, published_total climbing, publish_failed_total and dropped_total at 0.
6. Run the runbook verification end to end: stats /islands serving C{n} with a non-empty peer total (the join check, not just island count), /core-status healthy, heartbeat-fed endpoints unchanged, and a real client joining an island-C{n} LiveKit room.

Known gaps

• Cluster IDs are not unique beyond one process. C{n} counts from zero, so it resets on restart and collides across instances: after a restart C1 names a different crowd and gatekeeper maps it onto the previous C1's LiveKit room. Scope to server_id or a boot epoch. Live-voice-room correctness, not cosmetics.
• The §3.6 periodic re-publish sweep is unimplemented; delivery is at-most-once.
• Teleport bypass reads IsTeleport on the latest snapshot only, so a teleport followed by movement inside one pass goes through the debounce. Cross-realm teleports are always immediate.
• A subject moving mid-scan can be missed for one tick: AoI walks up to 25 cells lock-free, so the multi-cell read is not atomic. Pre-existing, unrelated to the write ordering above.
• engine.islands reports max_peers = 0 since clusters are uncapped, so GET /islands reads 0 where it read 100.
• Stats HTTP endpoints (§3.5) and engine.parcel_changes are iteration 2.

[decentraland-bot]

Code Review — PR #34: Peer Clustering over AoI Grid with NATS Feed

Thoroughly reviewed this large PR (~6,500 new lines across 88 files). The design is well-documented, the architecture is sound, and the test coverage is strong. One correctness bug needs fixing before merge.

Findings Summary

• P1 Critical: 1 — blocks merge
• P2 Minor: 3 — suggestions

---

P1 — Blocks Merge

1. PruneVanishedClusters modifies Dictionary during foreach enumeration

ClusterTracker.PruneVanishedClusters() iterates clusterRecords (a Dictionary<string, ClusterRecord>) and calls clusterRecords.Remove(clusterId) inside the loop. In .NET, this throws InvalidOperationException on the first removal.

The try/catch in RunPassLoop catches the exception and logs it, so the BackgroundService recovers on the next pass — but:

• Pruning never succeeds: every attempt to remove a vanished cluster throws before reaching the second entry.
• clusterRecords grows unbounded with stale cluster IDs, bounded only by the rate of cluster creation.
• An error is logged on every pass that has vanished clusters, creating log noise.

Fix: collect keys to prune into a temporary list, then remove outside the enumeration:

private void PruneVanishedClusters()
{
    if (clusterRecords.Count == components.Count) return;

    List<string>? toRemove = null;
    foreach ((string clusterId, ClusterRecord record) in clusterRecords)
    {
        if (record.LastLivePass != passNumber)
            (toRemove ??= []).Add(clusterId);
    }

    if (toRemove is null) return;
    foreach (string id in toRemove)
        clusterRecords.Remove(id);
}

---

P2 — Suggestions

2. No length validation on client-supplied realm names (RealmSpatialGrids.cs)

Realm names arrive from clients and are used as ConcurrentDictionary keys. The grid lifecycle is correctly bounded (one grid per peer, dropped when last occupant leaves), so the count is safe. However, there is no length cap on the realm string itself — a malicious client could send multi-megabyte realm names that persist as dictionary keys until the peer leaves. Consider capping realm name length at the intake boundary (e.g., TeleportHandler).

3. dropped counter may miscount on defensive eviction path (NatsPublisher.QueueChange)

In the eviction branch, dropped = true is set unconditionally after the changeOrder.TryDequeue, even when pendingChangeBySubject.Remove(evicted) returns false (i.e., the evicted subject was stale). Under normal operation, the lock invariant ensures that every subject in changeOrder is in pendingChangeBySubject, so the Remove should always succeed. The TryDequeueNext drain loop has the same defensive continue pattern, suggesting the author considered the possibility. In practice this appears correct, but the unconditional dropped = true means if the invariant ever breaks, the metric would overcount without an actual eviction.

4. Consider a contention metric for the shared RealmSpatialGrids write lock

All realms share a single Lock for writes. At sub-microsecond hold times and MaxPeers=4095, this is likely fine, but adding a contention metric would provide visibility under production load.

---

Security Review

• ✅ SanitizeBrokerUrl properly strips userinfo (credentials) from NATS URLs before logging, using Uri.TryCreate to extract only host and port.
• ✅ ConnStringRedaction redacts access_token= values from LiveKit connection strings via regex.
• ✅ No secrets or credentials in docker-compose.debug.yml or docker-compose.e2e.yml. E2E_NATS_URL is read from host env with a safe local default.
• ✅ AddNatsUrlAlias precedence is correct — the alias cannot override an explicit Nats__Url.
• ✅ SnapshotBoard.Publish return-type change (void → PeerSnapshot) is additive and backward-compatible for callers.

Architecture Notes

• The weighted union-find with path halving is a good fit — O(N+C) per pass, measured at ~395μs at the 4095-peer ceiling.
• Structural realm isolation via per-realm SpatialGrid instances eliminates per-candidate realm comparisons in the hot path.
• The coalescing outbox with per-feed supersede semantics (latest-wins for topology, per-peer for cluster_change) is well-designed.
• Fail-soft NATS integration: empty URL → stats-only mode; broker outage → outbox retention; no blocking on the hot path.
• The known gaps (cluster ID uniqueness, missing re-publish sweep, percolation at high density) are well-documented and appropriate to defer.

CI Status

All checks passing ✅ (Build linux-x64, Build osx-arm64, Build win-x64, Run tests).

---

Reviewed by Jarvis 🤖 · Requested by mikhail-dcl via GitHub

[decentraland-bot]

[P1] Bug: Dictionary modification during foreach enumeration.

This will throw InvalidOperationException in .NET whenever a cluster vanishes. The try/catch in RunPassLoop catches it, but pruning never completes — clusterRecords grows unbounded with stale entries.

Suggested fix: collect keys first, then remove:

private void PruneVanishedClusters()
{
    if (clusterRecords.Count == components.Count) return;

    List<string>? toRemove = null;
    foreach ((string clusterId, ClusterRecord record) in clusterRecords)
    {
        if (record.LastLivePass != passNumber)
            (toRemove ??= []).Add(clusterId);
    }

    if (toRemove is null) return;
    foreach (string id in toRemove)
        clusterRecords.Remove(id);
}