Membership Epochs

A fixed family preserves the membership it captured at entry. Membership epochs let the model end one cohort and deliberately capture the next.

import {
  Activity,
  Query,
  State,
  Statechart,
  States,
  Transitions,
} from "@motive/motive";
import * as Effect from "effect/Effect";
import { Schema } from "effect";

const StartQueue = Schema.TaggedStruct("StartQueue", {});

export const IndexJob = Activity.make("IndexJob", {
  input: Schema.String,
  success: Schema.String,
});

const Jobs = IndexJob.batch("Jobs");

export class JobQueue extends State.Compound<JobQueue>()("JobQueue", {
  queue: Schema.Array(Schema.String).pipe(
    Schema.withConstructorDefault(
      Effect.succeed(["alpha", "bravo", "charlie", "delta", "echo"]),
    ),
  ),
  done: Schema.Array(Schema.String).pipe(
    Schema.withConstructorDefault(Effect.succeed([])),
  ),
  epochs: Schema.Finite.pipe(Schema.withConstructorDefault(Effect.succeed(0))),
}) {
  static states = States.make(() => [Idle, Working, Drained]);
}

class Idle extends State.Atomic<Idle>()("Idle") {
  static transitions = Transitions.make(this, ({ on }) => [
    on(StartQueue, Working),
  ]);
}

class Working extends State.Atomic<Working>()("Working") {
  static transitions = Transitions.make(this, ({ invoke, on }) => [
    invoke(
      Jobs,
      Query.gen(function* () {
        return new Map(
          (yield* JobQueue).queue.slice(0, 2).map((job) => [job, job] as const),
        );
      }),
    ),
    on(Jobs.MemberDone).update(JobQueue, ({ event, target }) => ({
      queue: target.queue.filter((job) => job !== event.key),
      done: [...target.done, event.key],
    })),
    on(Jobs.Done, Drained).update(JobQueue, ({ target }) => ({
      epochs: target.epochs + 1,
    })),
  ]);
}

class Drained extends State.Atomic<Drained>()("Drained") {}

export const JobQueueChart = Statechart.make(JobQueue);

import {
  Activity,
  Query,
  State,
  Statechart,
  States,
  Transitions,
} from "@motive/motive";
import * as Effect from "effect/Effect";
import { Schema } from "effect";

const StartQueue = Schema.TaggedStruct("StartQueue", {});

export const IndexJob = Activity.make("IndexJob", {
  input: Schema.String,
  success: Schema.String,
});

const Jobs = IndexJob.batch("Jobs");

export class JobQueue extends State.Compound<JobQueue>()("JobQueue", {
  queue: Schema.Array(Schema.String).pipe(
    Schema.withConstructorDefault(
      Effect.succeed(["alpha", "bravo", "charlie", "delta", "echo"]),
    ),
  ),
  done: Schema.Array(Schema.String).pipe(
    Schema.withConstructorDefault(Effect.succeed([])),
  ),
  epochs: Schema.Finite.pipe(Schema.withConstructorDefault(Effect.succeed(0))),
}) {
  static states = States.make(() => [Idle, Working, Drained]);
}

class Idle extends State.Atomic<Idle>()("Idle") {
  static transitions = Transitions.make(this, ({ on }) => [
    on(StartQueue, Working),
  ]);
}

class Working extends State.Atomic<Working>()("Working") {
  static transitions = Transitions.make(this, ({ invoke, on }) => [
    invoke(
      Jobs,
      Query.gen(function* () {
        return new Map(
          (yield* JobQueue).queue.slice(0, 2).map((job) => [job, job] as const),
        );
      }),
    ),
    on(Jobs.MemberDone).update(JobQueue, ({ event, target }) => ({
      queue: target.queue.filter((job) => job !== event.key),
      done: [...target.done, event.key],
    })),
    on(Jobs.Done, Working)
      .reenter()
      .when(
        Query.gen(function* () {
          return (yield* JobQueue).queue.length > 0;
        }),
      )
      .update(JobQueue, ({ target }) => ({ epochs: target.epochs + 1 })),
    on(Jobs.Done, Drained).update(JobQueue, ({ target }) => ({
      epochs: target.epochs + 1,
    })),
  ]);
}

class Drained extends State.Atomic<Drained>()("Drained") {}

export const JobQueueChart = Statechart.make(JobQueue);

One arming defines one membership epoch

Jobs is a fixed Activity family:

const Jobs = IndexJob.batch("Jobs");

When Working enters, its entries query admits the first two queued keys:

invoke(
  Jobs,
  Query.gen(function* () {
    return new Map((yield* JobQueue).queue.slice(0, 2).map((job) => [job, job] as const));
  }),
);

That arming is one membership epoch. alpha and bravo belong to it even as their individual outcomes update other chart data. charlie, delta, and echo remain backlog; they were never members of this family occurrence.

Member progress may change the source collection

Each successful member removes its own key from JobQueue.queue:

on(Jobs.MemberDone).update(JobQueue, ({ event, target }) => ({
  queue: target.queue.filter((job) => job !== event.key),
  done: [...target.done, event.key],
}));

The queue changes from five jobs to three, but the active Jobs membership does not drift. A fixed family uses its committed epoch as authority until that occurrence retires.

This separation matters in recovery. The engine can redrive unresolved work from the captured membership without rerunning the producer against a queue that has already changed.

Aggregate completion closes the epoch

Jobs.Done means every member captured for this arming has settled. It does not mean the external backlog is empty, and it does not implicitly ask the entries query for more work.

The first frame targets Drained immediately. That makes the distinction visible: the family is honestly complete after two members, while three queued jobs remain.

Completion closes a lifecycle boundary. What happens next is an authored statechart decision.

Reentry arms the next epoch

When backlog remains, the second frame handles aggregate completion by reentering Working:

on(Jobs.Done, Working)
  .reenter()
  .when(
    Query.gen(function* () {
      return (yield* JobQueue).queue.length > 0;
    }),
  );

Exit retires the settled Jobs occurrence. Entry evaluates the producer again from committed JobQueue data and arms a fresh occurrence with a fresh captured membership.

The live run therefore advances through three epochs: alpha/bravo, charlie/delta, and echo. The family never changes membership while armed; the model changes which family occurrence is current.

Advance durable progress before rearming

The MemberDone update and aggregate completion belong to the same ordered statechart semantics. By the time Jobs.Done selects reentry, completed keys have already left queue. The next entries query therefore sees advanced durable progress.

If member completion failed to dequeue its key, the next epoch would capture that same job again. The family machinery cannot infer which domain collection represents completed work; the chart must make that progress explicit.

Give the empty epoch a terminal branch

An empty fixed family completes vacuously. Reentering Working unconditionally after Jobs.Done would allow an empty queue to arm an empty family, complete immediately, and reenter forever.

The guarded reentry handles the non-empty case. A following transition to Drained handles the complement:

on(Jobs.Done, Drained);

The pair is both total and terminating: continue while work remains; leave when it does not.

Epochs and open families solve different problems

Use repeated .batch epochs when each cohort must remain fixed after admission: bounded batches, transaction groups, rollout waves, or recipients captured for one notification. Use .each when membership should follow desired model data continuously while one owner remains active.

An epoch is not merely a concurrency trick. Here slice(0, 2) also bounds peak work at two, but the deeper guarantee is cohort identity: the exact two jobs admitted together remain the authority for that arming and its aggregate outcome.

Test the epoch boundary

1. Verify the entries producer runs once for each Working incarnation, not once per member.
2. Complete members in different orders and verify the captured cohort does not drift.
3. Verify durable dequeue commits before the next epoch's producer runs.
4. Recover midway through an epoch and verify only its unresolved captured members are redriven.
5. Start with an empty queue and verify the chart reaches Drained without an eventless loop.

Next, member outcomes and aggregate barriers expose what a fixed family remembers about every settled key and how the chart observes individual progress versus whole-cohort completion.