Sharded Daemon Process

Manchmal brauchst du N Hintergrund-Worker, verteilt über den Cluster, jeder mit stabiler Identität:

• Ein Worker pro Arbeits-Partition (denke Kafka-Consumer-Group-Muster, aber innerhalb des Clusters).
• N geplante Koordinatoren, einer pro Zeitzone, einer pro Region.
• N Hintergrund-Reconciler, jeder besitzt eine Scheibe der Work-IDs.

Ein Singleton gibt dir einen davon. Sharding gibt dir einen pro Entity-ID. ShardedDaemonProcess gibt dir genau N, indiziert 0..N-1, gleichmäßig über den Cluster verteilt, automatisch bei Failover neu gespawnt.

Ein minimales Beispiel

import { ActorSystem, Actor } from 'actor-ts';
import { Cluster, ClusterOptions } from 'actor-ts/cluster';
import { ShardedDaemonProcess, ShardedDaemonProcessOptions } from 'actor-ts/cluster';

class Reconciler extends Actor<{ kind: 'check' }> {
  constructor(public readonly index: number) { super(); }

  override preStart(): void {
    this.log.info(`Reconciler #${this.index} startet`);
    // plane periodische Prüfung, z. B.:
    this.context.timers.startTimerWithFixedDelay(
      'check', { kind: 'check' }, 60_000,
    );
  }

  override onReceive(message: { kind: 'check' }): void {
    this.log.info(`Reconciler #${this.index} tickt`);
    // ... Per-Index-Arbeit
  }
}

const system  = ActorSystem.create('my-app');
const clusterOptions = ClusterOptions.create()
  .withHost(host)
  .withPort(port)
  .withSeeds(seeds);
const cluster = await Cluster.join(system, clusterOptions);

const shardedDaemonProcessOptions = ShardedDaemonProcessOptions.create()
  .withName('reconciler')
  .withNumDaemons(8)
  .withActorFor((i) => () => new Reconciler(i));
const handle = ShardedDaemonProcess.init(system, cluster,
  shardedDaemonProcessOptions);

// Jeder der 8 Daemons startet automatisch.  Kein weiteres Setup nötig.

// Optional: schicke eine Nachricht an einen spezifischen Index
handle.tell(3, { kind: 'check' });

Drei Dinge passieren bei init:

1. Das Framework registriert einen sharded Entity-Typ (daemon-reconciler).
2. Jede Entity-ID ist einfach der Index des Daemons als String ('0', '1', …, '7').
3. Interne “Wake-up”-Nachrichten werden an jeden Index geschickt, was die Sharding-Maschinerie zwingt, die Entity-Actors auf den vom Koordinator gewählten Nodes zu materialisieren.

Danach laufen alle 8 Daemons, verteilt über Cluster-Nodes, jeder mit seinem stabilen Index sichtbar via Konstruktor-Argument.

Was der Handle dir gibt

interface ShardedDaemonProcessHandle<T> {
  readonly region: ActorRef<DaemonEnvelope<T>>;
  tell(index: number, message: T): void;
  stop(): void;
}

• region — der darunterliegende Sharding-Region-Ref. Nachrichten, die hier gesendet werden, müssen in { index, body }-Envelopes verpackt sein; nutze stattdessen handle.tell.
• tell(index, message) — Convenience zum Senden an Daemon i.
• stop() — stoppt den Liveness-Heartbeat und die Cluster-Subscription. Stoppt die Daemons selbst nicht — sie laufen weiter, bis der Cluster heruntergefahren wird.

Wie es auf Sharding aufbaut

ShardedDaemonProcess ist ein dünner Wrapper um ClusterSharding:

• typeName: 'daemon-<name>'.
• extractEntityId: env => env.index.toString().
• rememberEntities: true — sodass ein Node-Ausfall den Daemon woanders neu spawnt.
• allocationStrategy: LeastShardAllocationStrategy per Default — verteilt Daemons gleichmäßig.

Du könntest all das manuell mit ClusterSharding.start schreiben. Der Wrapper existiert, um:

• Dir Boilerplate zu sparen.
• Bei Init automatisch Wake-up-Nachrichten zu senden, damit Daemons eager materialisieren.
• LeaderChanged / MemberRemoved zu abonnieren und Wake-ups erneut zu senden, was sicherstellt, dass Daemons, die laufen sollten, auch tatsächlich laufen.
• Das livenessIntervalMs-Sicherheitsnetz bereitzustellen.

Liveness-Heartbeat

const shardedDaemonProcessOptions = ShardedDaemonProcessOptions.create()
  .withName('reconciler')
  .withNumDaemons(8)
  .withActorFor((i) => () => new Reconciler(i))
  .withLivenessIntervalMs(30_000);
ShardedDaemonProcess.init(system, cluster,
  shardedDaemonProcessOptions);   // Default — pingt jeden Index alle 30s

Periodisch (Default alle 30 s) schickt der Wrapper eine synthetische “Wake-up” an jeden Index. Wenn ein Daemon für irgendeinen Index unerwartet gestoppt wurde (ein Node, der geht + eine kurze Partition genau im falschen Moment), bringt das Wake-up die Sharding-Maschinerie dazu, ihn auf einem aktuellen Up-Mitglied neu zu spawnen.

Das ist doppelt gemoppelt — das meiste Failover regelt Sharding mit seinen eigenen Cluster-Event-Subscriptions. Der Liveness-Ping deckt den Edge Case ab, in dem ein Event verpasst wurde.

Auf 0 setzen deaktiviert es. Nützlich in Tests, in denen du Cluster-Events deterministisch kontrollierst.

Konfiguration

type ShardedDaemonProcessOptionsType<T> = {
  name:                 string;          // eindeutiger Daemon-Set-Identifier
  numDaemons:           number;          // Gesamtdaemons über den Cluster
  actorFor:             (i: number) => ActorClassOrFactory<T>;
  role?:                string;          // auf spezifische Nodes beschränken
  livenessIntervalMs?:  number;          // Sicherheitsnetz-Ping-Takt; Default 30s
};

Feld	Was
name	Ein eindeutiger Daemon-Set-Name. Zwei init-Aufrufe mit demselben Namen kollidieren.
numDaemons	Wie viele Daemons laufen sollen. Bei Init festgelegt; kann ohne Neustart nicht geändert werden und muss auf jedem Node identisch sein — der Wert wird zum numShards des Sharding-Typs, und der Koordinator verweigert einen Node, der abweicht (siehe numShards ist eine clusterweite Vereinbarung). Ein Rolling Deploy, der ihn ändert, lässt die noch nicht aktualisierten Nodes registriert und die aktualisierten verweigert, bis der letzte durch ist.
actorFor(i)	Gibt die Actor-Klasse bzw. Factory für Daemon i zurück. Nutze i, um Per-Index-Logik zu unterscheiden.
role	Nur Nodes mit dieser Rolle hosten Daemons (durchgereicht an Sharding).
livenessIntervalMs	Sicherheitsnetz-Ping-Takt; Default 30s.

Wann zugreifen

Drei gute Anwendungen:

1. Partitionierte Hintergrundarbeit — N Reconciliation-Jobs, jeder besitzt eine Scheibe der IDs. Modulo oder Hash, um Arbeit auf den Daemon-Index abzubilden.
2. Verteilte Scheduler — ein Per-Zeitzone-Scheduler, ein Per-Region-Quota-Replenisher. Der Index jedes Daemons mappt auf seinen Scope.
3. Stream-Processing-Groups — bei Integration mit Kafka-/Broker-Actors kann jeder Daemon ein Consumer-Group-Mitglied sein.

Wann NICHT verwenden

Ein Daemon = nimm einen Singleton

const daemonOptions = ShardedDaemonProcessOptions.create().withNumDaemons(1)/* ... */;
ShardedDaemonProcess.init(..., daemonOptions);   // ✗

Für numDaemons: 1 nimm ClusterSingleton — er hat explizite Semantik für “genau einer clusterweit” (Wahl, Lease etc.), während Sharded-Daemon mit N=1 fast ein Singleton ist, aber die expliziten Garantien verliert.

Viele Daemons = nimm Sharding direkt

const daemonOptions = ShardedDaemonProcessOptions.create().withNumDaemons(100_000)/* ... */;
ShardedDaemonProcess.init(..., daemonOptions);   // ✗

Sharded-Daemon startet jeden Index eager. Für großes N würdest du Tausende Actors zur Init-Zeit spawnen — eine donnernde Herde. Nutze stattdessen normales Sharding mit lazy Entity-Erstellung (Entities existieren erst nach der ersten Nachricht).

Das Verhalten jedes Daemons muss deterministisch nach Index sein

actorFor: (i) => () => new Reconciler(Math.random()),
// ↑ Random ist zwischen Leader-Wechseln unterschiedlich; Daemon 3 auf
//   node-A und Daemon 3 auf node-B nach Failover verhalten sich anders

Die Factory muss dasselbe Verhalten für denselben Index produzieren — sonst gibt dir ein Failover einen anderen Daemon 3 als vorher. Nutze i (und etwaige externe Lookups, die per i gekeyt sind) konsistent.

Vergleich

Werkzeug	Was
ClusterSingleton	Genau 1 Actor clusterweit.
ShardedDaemonProcess	Genau N indizierte Actors, verteilt.
ClusterSharding	Ein Actor pro Key (potenziell unendlich).
ClusterRouter	Routet Nachrichten über vorab existierende Actors auf jedem Node.

“Das richtige Werkzeug nach Kardinalität”: 1 → Singleton, feste N → Daemon-Process, viele (eine pro Business-Entity) → Sharding, “jeder Node” → Cluster-Router.

Wohin als Nächstes

• Sharding-Überblick — das Fundament, auf dem es aufbaut.
• Singleton-Überblick — für N=1.
• Allokationsstrategie — was die Daemons platziert.
• Remember Entities — was sie über Koordinator-Failover hinweg überleben lässt.

Die ShardedDaemonProcess API-Referenz deckt die vollständige Oberfläche ab.