Adding a Reactor

1. Before You Start

Read Signal & Sweep Framework Architecture first. This page assumes you know what a signal, reactor, sweep and aggregate key are, and it will not re-explain them.

Then check Reactor Catalogue — your concern may already be described there, with a signal type and debounce window chosen.

1.1. Do you need a reactor at all?

Situation Use

Work must happen soon after a domain change, and the change goes through application code

A reactor with a signal handler

Work must happen on a timetable, or must catch changes that bypass application code

A reactor with a sweep

Both — needs to be quick, but must never be missed

A reactor with both. This is the common case.

Work must complete before the caller’s request returns

Not a reactor. Do it synchronously.

Work is a multi-step, long-running business process with escalation and human steps

Not a reactor on its own — a reactor that starts a Fluxnova process

If your work must be transactionally atomic with the domain change, a reactor is the wrong tool. Signals are committed with the caller but processed later, in a different transaction.

2. Step 1 — Decide the Shape

Answer four questions before writing code.

What is the unit of work? This becomes the aggregate key. If you recalculate a whole series, the unit is the series and the key is series:{id} — not result:{id}, which would give you one recalculation per result.

How stale can the result be? This becomes the debounce window. Expensive work triggered by bursts wants a long window. A customer-facing message wants zero — squashing two payment confirmations means one customer never hears from you.

Can it run twice safely? It must. Sweeps re-run on schedule, signals are retried, and an operator can trigger a sweep by hand. If running twice causes damage, add a guard in your own code — the framework will not do it for you.

What happens if it never runs? If the answer is "someone notices eventually", a signal handler alone is fine. If the answer is "silent data corruption", you need a sweep as well.

3. Step 2 — Implement the Interface

Create the class in za.co.idealogic.event.admin.trigger.reactor:

@Component
public class ExampleReactor implements Reactor {

    private static final Logger log = LoggerFactory.getLogger(ExampleReactor.class);
    private static final String NAME = "example";

    private final SignalSweepProperties properties;
    private final ExampleDomainService domainService;

    public ExampleReactor(SignalSweepProperties properties,
                          ExampleDomainService domainService) {
        this.properties = properties;
        this.domainService = domainService;
    }

    @Override
    public String getName() {
        return NAME;   // matches ems.signal-sweep.reactors.example
    }

    @Override
    public Set<String> getHandledSignalTypes() {
        return Set.of("REGISTRATION_COMPLETED");
    }

    @Override
    public void handleSignal(DomainSignal signal) {
        Long orderId = signal.getEntityId();
        log.debug("Reactor {} handling signal {} for order {}",
            NAME, signal.getId(), orderId);
        domainService.doTheWork(orderId);
    }

    @Override
    public void sweep() {
        // find outstanding work by inspecting domain state, then process it
    }

    @Override
    public String getSweepCron() {
        return properties.getReactor(NAME).getSweepCron();
    }
}

Being a Spring bean is all the registration there is. There is no central list to edit.

For a sweep-only reactor return Set.of() from getHandledSignalTypes(). For a signal-only reactor return null from getSweepCron() and leave sweep() empty.

3.1. Rules for handleSignal

  • Read current entity state, not the payload, wherever correctness depends on it. Squashing overwrites the payload with the most recent change, so it is a hint, not a history.

  • Throw on failure. That is how you tell the framework to retry. Do not catch and swallow — a silently failing reactor is invisible.

  • Throw only for your own concern. Other reactors on the same signal are tracked separately and are unaffected by your failure.

  • Keep it short. You are on the Signal Master’s poll cycle. Long work belongs in a sweep, or should be broken into per-entity transactions.

3.2. Rules for sweep

  • Idempotent, always.

  • Bound the work. Page or batch rather than loading every candidate row.

  • Isolate failures per item. Process each entity in its own transaction so one bad row does not abort the batch. This is the same pattern the existing per-event batch assignment uses.

  • Log counters, not just completion: candidates found, processed, skipped, failed. These become the reactor’s operational signal.

4. Step 3 — Configure

Add the reactor’s subtree to application.yml:

ems:
  signal-sweep:
    reactors:
      example:
        enabled: true
        sweep-cron: "0 */10 * * * *"
        organisations: []               # tenants this reactor may act for; empty means NONE
        # reactor-specific settings belong here too
        dry-run: false
        max-age-days: 30

organisations is empty by default and empty means none, so a new reactor is inert until an environment names its tenants. That is deliberate — see Organisation Scoping. Per-environment values belong in the deployment manifest, never in a profile file: application-prod.yml reaches stage and both production lines alike.

If your reactor touches nothing belonging to a tenant — it maintains framework tables only — override isOrganisationScoped() to return false and say why inline. Leaving the default of true on such a reactor would keep it permanently unable to run.

If you introduce a new signal type, give it a debounce window:

ems:
  signal-sweep:
    debounce:
      MY_NEW_SIGNAL: 30s

Profile defaults:

  • application-dev.yml — off. A developer’s machine must not sweep a shared database or send email.

  • application-prod.yml — on, except anything destructive, which starts disabled with a dry-run flag.

Anything that deletes data, sends a message, or calls an external system starts disabled and is enabled deliberately once verified.

5. Step 4 — Emit the Signal

Find the point in the domain service where the change is committed and emit there:

order.setStatus(OrderStatus.PAID);
orderRepository.save(order);

signalPublisher.emit(
    "PAYMENT_COMPLETED",           // signal type
    "Order",                       // entity type
    order.getId(),                 // entity id
    "order:" + order.getId(),      // aggregate key
    order.getOrganisationId(),     // organisation context, nullable
    "{\"amount\":" + amount + "}"  // optional JSON payload, nullable
);

Three things to get right:

  1. Inside the caller’s transaction, after the change. If the transaction rolls back the signal must vanish with it. emit is declared Propagation.MANDATORY, so calling it without a transaction throws rather than quietly creating an orphan signal. If you hit that exception, fix the call site — do not widen the propagation.

  2. Not from an AFTER_COMMIT listener or a new transaction. That reintroduces exactly the orphan-signal problem the MANDATORY propagation exists to prevent.

  3. Aggregate key at the reactor’s granularity, matching the convention in the catalogue.

Emitting is cheap and safe from any pod, and it is a single atomic statement — there is no race to handle and no exception to catch. Only processing is restricted to the leader.

Note that for a signal type configured with a zero debounce window, the aggregate key you pass is deliberately not stored — see the architecture page. Pass it anyway so the call site reads uniformly; the publisher decides based on configuration.

6. Step 5 — Test

Reactors are testable without the rest of the framework running.

Signal handling — construct a DomainSignal and call handleSignal directly. Assert the domain effect. Then call it twice and assert the effect is the same, proving idempotency.

Sweep — seed domain rows in each relevant state, run sweep(), and assert only the intended rows were touched. Include boundary cases explicitly: a row just inside the window, one just outside, and one in a state the sweep must never touch.

Emission — call the domain service and assert a row appears in signal_domain with the expected type, aggregate key and process_after. Then assert that a rolled-back transaction leaves none.

Squashing — emit twice in quick succession with the same key and assert there is one row with process_after pushed forward, not two rows.

Retry — make the reactor throw, run the processor, and assert signal_reactor_status shows FAILED with a next_retry_at. Assert that a second reactor on the same signal is unaffected.

Two cautions specific to this codebase:

  • Scheduling is disabled in test profiles. Drive the processor and sweeps directly from the test rather than waiting for a tick.

  • admin-service has no CI build or test gate on pull requests. Local mvn verify is the only check that runs. Run it before you push, and do not read a green pull request as evidence that anything was tested.

7. Step 6 — Verify in a Running System

With the service running, use the admin API (ROLE_ADMIN):

# Is my reactor registered, and is it enabled?
curl -s .../api/admin/signals/reactors

# Turn it on for this session without touching config
curl -X PUT .../api/admin/signals/reactors/example/enabled \
  -H 'Content-Type: application/json' -d '{"enabled": true}'

# Run the sweep now instead of waiting for the cron
curl -X POST .../api/admin/signals/reactors/example/sweep

# What happened to a specific signal, per reactor?
curl -s .../api/admin/signals/{id}

See Management API Access for authentication against deployed environments.

8. Debugging

Symptom Where to look

Signal never appears

The emit is outside the transaction, or the transaction rolled back. Check the domain change actually committed.

Signal appears but stays PENDING

process_after is in the future (debounce still sliding), or nothing holds the SIGNAL_MASTER lease, or the processor is disabled. Check GET /signals/leaders and GET /signals/status.

Signal goes straight to COMPLETED, nothing happened

No reactor is registered for that signal type. Check the spelling in getHandledSignalTypes() — it is a string match.

Reactor never fires but the signal completes

The reactor is disabled. Check effective state in GET /signals/reactors.

Signal stuck in PROCESSING

A reactor hung, or the processor died mid-flight. GET /signals/stuck.

Sweep never runs

getSweepCron() returned null, the cron expression is invalid, or this pod does not hold SWEEP_MASTER.

Work happens twice

Your reactor is not idempotent. The framework will retry, re-sweep and failover — it does not promise exactly-once.

9. Common Pitfalls

Treating the payload as authoritative. It is overwritten on squash. Read the entity.

Keying too finely. participant:{id} when the work is per-series gives you no coalescing at all.

Debouncing a communication. A non-zero window on a signal type that maps one-to-one to a customer message will eventually swallow one.

Assuming exactly-once. Retries, sweeps and failover all mean at-least-once. Design for it.

Catching your own exceptions. Swallowing a failure marks the reactor COMPLETED and the work is never retried.

Enabling a destructive reactor by default. Start disabled with a dry-run, prove the candidate set on real data, then enable.

Writing a sweep query that ignores the organisation set. The framework filters signal routing for you, and refuses to start a scoped sweep whose set is empty — but it cannot inspect your query. A sweep that selects across all tenants is a defect the framework cannot catch, so cover it with a test.

Overriding isOrganisationScoped() to false to make a reactor run. It is a statement that the reactor touches no tenant data, not a way past an empty list. If the reactor reads a shared table, the answer is to name its organisations.

Editing the Reactor interface. Two workstreams code against it. Changing it is a joint decision, not a unilateral edit.

10. Related Documentation