[Feature] External shard allocation + manual rebalance API

[pathosDev]

Size / Priority

• Size: M
• Reference: Akka Cluster Sharding — External Shard Allocation.

Rationale

Built-in AllocationStrategy impls cover ~80% of needs:

• HashAllocationStrategy (deterministic by shardId % N).
• LeastShardAllocationStrategy (load-balance by current count).

Production users sometimes need to influence shard placement based on signals outside the sharding subsystem's view:

• Capacity-aware — node X has 90% RAM used; don't place new shards there.
• Latency-aware — node Y is in DC us-east; place shards for us-east-tenants there.
• Business-rule-aware — tenant T must run on node Z (GDPR data residency).
• Operator-controlled — manual "move shard 42 from node A to node B" via REST endpoint (canary, evacuation, debugging).

Today these require sub-classing AllocationStrategy and recompiling. The framework should expose:

1. Pluggable strategy hook: takes an "external signal" (opaque object) and returns allocations.
2. Manual rebalance API: POST /actor-ts/cluster/shards/rebalance { shards: [{id, target}] } — operator initiates explicit moves.

Reference: what Akka does

val typeName = "Counter"
val externalShardAllocationStrategy = new ExternalShardAllocationStrategy(
  system, typeName, defaultStrategy = new LeastShardAllocationStrategy(...)
)
ClusterSharding(system).init(Entity(typeName)(...).withAllocationStrategy(externalShardAllocationStrategy))

// Operator pushes an explicit allocation:
val client = ExternalShardAllocation(system).clientFor(typeName)
client.updateShardLocation(shardId = "42", location = "akka://sys@host:port")

// Akka exposes an admin API:
// POST /cluster/shards/typeName/locations  { "shardId": "42", "address": "akka://..." }

The default strategy is the fallback; the external strategy overrides for shards that have been explicitly placed.

Design sketch — actor-ts equivalent

1. New AllocationStrategy.external() factory.

export interface ExternalAllocationState {
  /** Map<shardId → preferred owner address>; consulted before the fallback strategy. */
  readonly explicitPlacements: ReadonlyMap<number, NodeAddress>;
}

export function externalAllocationStrategy(
  fallback: AllocationStrategy,
  getState: () => ExternalAllocationState,
): AllocationStrategy;

Behaviour:

• allocate(shardId, candidates):
  1. If explicitPlacements.has(shardId): verify the target is in candidates; if so, return it. Otherwise fall through to fallback.
  2. Else: delegate to fallback.
• rebalance(currentShards, candidates, rebalanceInProgress):
  1. Find shards whose actual owner doesn't match explicitPlacements.
  2. Add them to the rebalance set (so coordinator re-allocates → external strategy picks the explicit target).
  3. Also union the fallback's rebalance recommendations for any shards without explicit placement.

2. REST endpoint for operator-initiated placement.

POST /actor-ts/cluster/shards/{typeName}/placements
Body: { "shardId": 42, "target": "actor-ts://app@10.0.0.5:7777" }

Persists to a small per-type allocation state in the coordinator. State is gossiped along with normal sharding state.

GET /actor-ts/cluster/shards/{typeName}/placements
→ { "placements": [{ "shardId": 42, "target": "actor-ts://..." }, ...] }

DELETE /actor-ts/cluster/shards/{typeName}/placements/{shardId}
→ removes the explicit placement; shard reverts to fallback strategy on next rebalance.

3. Manual rebalance API.

POST /actor-ts/cluster/shards/{typeName}/rebalance
Body: { "shardIds": [42, 43, 44] }

Forces the coordinator to re-evaluate those shards (HandOff + Allocate). With an external placement set, this is how operators move shards "right now" without waiting for the next rebalance tick.

4. Programmatic API.

export class ExternalAllocationController {
  constructor(private readonly cluster: Cluster, private readonly typeName: string);

  /** Set or update an explicit placement.  Persists + gossips. */
  async setPlacement(shardId: number, target: NodeAddress): Promise<void>;

  /** Remove an explicit placement.  Shard reverts to fallback on next rebalance. */
  async removePlacement(shardId: number): Promise<void>;

  /** Get current explicit placements. */
  async getPlacements(): Promise<ReadonlyMap<number, NodeAddress>>;

  /** Force re-evaluation of these shards (HandOff + Allocate). */
  async rebalanceNow(shardIds: ReadonlyArray<number>): Promise<void>;
}

Integration with existing actor-ts subsystems

• ShardCoordinator: holds the explicitPlacements map; gossips changes to other regions; consulted on every allocation decision.
• AllocationStrategy: new externalAllocationStrategy() factory; existing strategies (HashAllocationStrategy, LeastShardAllocationStrategy) are unchanged.
• HTTP management endpoint (Cluster-management HTTP endpoints — extended (shards, leave-and-shutdown, metrics) #56): depends on it for the REST API.
• Gossip: ShardingProtocol.PlacementUpdate message added; piggy-backs on existing sharding gossip.

Out of scope / non-goals

• Cross-cluster placement (move a shard to a different cluster): out of scope; that's a cluster-mesh concern, not external allocation.
• Latency-aware automatic placement: out of scope; the framework provides the hook + the API, not the latency-measurement logic.
• Capacity-aware automatic placement: same — providing the hook, not the measurement.
• Persistent external state across coordinator restart: phase 1 keeps placements in coordinator memory + gossip; if every region restarts simultaneously, placements are lost. Phase 2 (separate ticket) could persist to the journal.

Open design questions

1. What if the target node is unreachable? External placement says "shard 42 → node B"; node B is unreachable. Options:
   • Block the shard (don't allocate anywhere) until B recovers.
   • Fall back to default strategy (current sketch).
   • Mark the placement as "pending" and allocate elsewhere temporarily.
     Recommend: fall back + warn; explicit placement is a preference, not a constraint.
2. REST API auth: management endpoints currently rely on user-supplied middleware. Document the requirement.
3. State propagation latency: how fast does a setPlacement from operator's REST call propagate to all regions? Currently sharding gossip is ~1s. For canary scenarios this is fine; for emergency evacuation, faster.
4. Conflict resolution: two operators set conflicting placements for the same shard. Last-write-wins (timestamp-based) vs. coordinator-as-tie-breaker. LWW is simpler.

Test plan

1. Explicit placement honored — setPlacement(42, nodeB) then trigger allocation; shard 42 lands on B.
2. Fallback strategy on missing placement — shards without explicit placement use the fallback (e.g. LeastShard).
3. Removed placement — removePlacement(42); next rebalance moves shard 42 according to fallback.
4. Target unreachable — setPlacement(42, nodeB); nodeB unreachable; shard allocated via fallback + warning logged.
5. Manual rebalance — rebalanceNow([42]); coordinator emits HandOff + Allocate for shard 42 within one tick.
6. REST POST — POST /actor-ts/cluster/shards/Counter/placements body parsed correctly + persists.
7. REST GET — returns current placements.
8. REST DELETE — removes placement.
9. Gossip propagation — setPlacement on region A reaches region B within one gossip interval.
10. Coordinator restart — placements lost (documented phase-1 limit).

Acceptance criteria

• externalAllocationStrategy(fallback, getState) factory exported.
• ExternalAllocationController class exported.
• REST endpoints POST/GET/DELETE /actor-ts/cluster/shards/{typeName}/placements + POST .../rebalance.
• ShardCoordinator gossips placement changes.
• Fallback strategy used when target unreachable, with warning log.
• Documentation: "When to use external allocation" decision guide + canary example.
• Test suite covers all 10 cases.
• CHANGELOG entry under "Sharding: external allocation strategy + manual rebalance API".