Signal & Sweep Framework Architecture

1. Overview

Signal & Sweep is the asynchronous trigger framework in admin-service. It replaces the pattern of adding a bespoke @Scheduled service for every new background concern with one mechanism that all concerns share.

The framework answers two questions for any piece of deferred work:

  • Something changed — react to it. A domain service emits a signal; the framework routes it to whichever reactors care.

  • Nothing told us, but check anyway. Each reactor declares a sweep — a periodic reconciliation scan that finds work the signals missed.

The term trigger is used throughout instead of "event", because "event" already means a sporting event in the EMS domain. A DomainSignal is not a Spring ApplicationEvent.

This document is reactor-agnostic: it describes the framework only. You should be able to implement a new reactor from this page alone. For the catalogue of concrete reactors and the signal types they consume, see Reactor Catalogue. For a step-by-step cookbook, see Adding a Reactor.

2. Core Concepts

Concept Definition

Signal

A persistent record that something happened in the domain. Emitted fire-and-forget by application code, stored in signal_domain, consumed by the framework. A signal is a notification, not a command — it says what changed, not what to do about it.

Reactor

A Spring bean declaring (a) which signal types it handles and (b) a sweep method plus schedule. One processing concern implements one Reactor. A reactor may be signal-only, sweep-only, or both.

Sweep

A periodic run of a reactor’s reconciliation logic. Sweeps find work by inspecting domain state — timestamps, null columns, status flags — rather than by reading signals. Sweeps must be idempotent.

Debounce (squash-on-emit)

Multiple signals for the same concern and entity inside a time window collapse into one. Implemented at emit time by pushing an existing pending signal’s process_after forward instead of inserting a duplicate.

Aggregate key

The string that defines "the same concern and entity" for squashing — for example series:8 or order:1423. Two signals of the same type with the same aggregate key are the same work.

2.1. Why both signals and sweeps

Signals alone are not enough:

  • Not every change goes through application code — SQL scripts, batch imports and external callbacks all bypass the emit point.

  • If the application dies after the domain change commits but before the signal does, the work is lost.

  • Some concerns have no discrete trigger at all; they are purely time-based.

Sweeps alone are not enough:

  • For time-sensitive reactions such as a confirmation email, waiting for the next sweep is too slow.

  • Sweep frequency trades responsiveness against database load. Signals remove that trade-off.

Together they are belt and braces: signals give speed, sweeps give correctness. A reactor that must never miss work should implement both, and its sweep should be able to find everything its signal handler would have done.

3. Signal Lifecycle

signal-lifecycle

A signal is matured when process_after ⇐ NOW(). The processor never sees a signal before its debounce window has elapsed.

A signal with no registered reactor for its type is picked up and marked COMPLETED immediately, having done nothing. This is deliberate and useful: it lets a signal type be emitted and observed in production before any reactor acts on it.

4. Data Model

Three tables, all in the database module. None of them are Org-Scoped — they are system service tables and do not participate in the security model. See Security classification.

All three share the signal_ prefix so they group together in database tools. Note that the table name and the JPA entity name deliberately diverge for the first one: the table is signal_domain (for grouping), the entity is DomainSignal (because that is what it is). Do not "fix" either to match the other.

4.1. signal_domain

CREATE TABLE signal_domain (
    id              BIGINT AUTO_INCREMENT PRIMARY KEY,
    signal_type     VARCHAR(100) NOT NULL,      -- e.g. 'RESULT_SET_CHANGED'
    entity_type     VARCHAR(100),               -- e.g. 'ResultSet', 'Order'
    entity_id       BIGINT,                     -- PK of the changed entity
    aggregate_key   VARCHAR(200),               -- squash key, e.g. 'series:8'
    payload         JSON,                       -- optional context
    organisation_id BIGINT,                     -- context/filtering only
    status          VARCHAR(20) DEFAULT 'PENDING',
    created_at      TIMESTAMP DEFAULT NOW(),
    process_after   TIMESTAMP NOT NULL,         -- debounce: not before this time
    updated_at      TIMESTAMP,                  -- last squash time
    processed_at    TIMESTAMP,
    error_message   VARCHAR(2000),
    retry_count     INT DEFAULT 0,
    max_retries     INT DEFAULT 3,
    next_retry_at   TIMESTAMP,                  -- exponential backoff
    INDEX idx_ready (status, process_after),
    INDEX idx_squash (signal_type, aggregate_key, status)
);
Column Notes

process_after

The debounce deadline. The processor selects only status = 'PENDING' AND process_after ⇐ NOW(). Squashing pushes this forward, producing a sliding window.

aggregate_key

Defines squash identity together with signal_type. Conventions are per-reactor — see Reactor Catalogue.

payload

Optional context to save a re-query. Replaced by the latest change on squash, so never treat it as a change history. Reactors should read current entity state, not the payload, whenever correctness depends on it.

organisation_id

Nullable. For filtering in admin queries only — not a security boundary.

updated_at

When the signal was last squashed. Useful diagnostically: a large gap from created_at means a long burst of activity.

next_retry_at

Exponential backoff. The processor skips signals whose next_retry_at is still in the future.

4.2. signal_reactor_status

One signal type may be consumed by several reactors. A single status column on signal_domain cannot express "reactor A succeeded, reactor B failed", so delivery is tracked per reactor.

CREATE TABLE signal_reactor_status (
    id              BIGINT AUTO_INCREMENT PRIMARY KEY,
    signal_id       BIGINT NOT NULL,
    reactor_name    VARCHAR(100) NOT NULL,
    status          VARCHAR(20) NOT NULL,    -- PENDING, COMPLETED, FAILED, DEAD
    processed_at    TIMESTAMP,
    error_message   VARCHAR(2000),
    retry_count     INT DEFAULT 0,
    UNIQUE KEY uk_signal_reactor (signal_id, reactor_name),
    FOREIGN KEY (signal_id) REFERENCES signal_domain(id) ON DELETE CASCADE
);

The natural key is (signal_id, reactor_name), but it is enforced as a unique index over a surrogate id rather than used as a composite primary key. This follows the codebase convention — no domain entity uses @IdClass or @EmbeddedId — and keeps Spring Data repositories straightforward. The uniqueness guarantee is identical.

This table exists so that a retry re-invokes only the reactors that failed. Without it, retrying a signal because reactor B timed out would run reactor A a second time — wasteful when A is expensive, and dangerous when A has a non-idempotent side effect such as sending an email.

The signal’s overall status is derived from its reactor rows:

Reactor rows Signal status

All COMPLETED

COMPLETED

Any FAILED with retries remaining

stays PROCESSING

All either COMPLETED or DEAD

COMPLETED, with a warning logged for each DEAD reactor

4.3. signal_sweep_lease

CREATE TABLE signal_sweep_lease (
    lease_name      VARCHAR(50) PRIMARY KEY,   -- 'SIGNAL_MASTER' or 'SWEEP_MASTER'
    holder_id       VARCHAR(100) NOT NULL,     -- pod identifier
    acquired_at     TIMESTAMP NOT NULL,
    expires_at      TIMESTAMP NOT NULL,
    INDEX idx_expires (expires_at)
);

4.4. Security classification

All three tables are system service tables, not user-facing entities:

  • No OrganizationalSecured interface, no SecurityDimensionService integration.

  • organisation_id on signal_domain is nullable, and is used for admin filtering and for reactor scoping — see Organisation Scoping. It is not part of the security model: no reactor derives a caller’s rights from it.

  • Access to signal data is via the admin API, which is ROLE_ADMIN only.

See Entity Classification for how this differs from Org-Scoped entities.

5. Debounce: Squash on Emit

Debounce happens when a signal is emitted, not when it is processed. This keeps the processor simple — it only ever sees work that is ready.

SignalPublisher.emit() does:

  1. Look up the debounce window configured for this signal_type.

  2. Issue one INSERT …​ ON DUPLICATE KEY UPDATE against signal_domain. If a PENDING signal already exists for the same signal_type + aggregate_key, its process_after moves to now + window, its payload is replaced and updated_at is set. Otherwise a new signal is inserted.

One statement, and no lock contention beyond the single row being written.

5.1. Why one statement rather than update-then-insert

The obvious implementation is two statements: try to squash, and insert if there was nothing to squash into. Two threads can then both find nothing and both insert.

uk_signal_domain_pending (see Data Model) turns that race into a constraint violation rather than a silent duplicate, which is an improvement — but not one that can be caught and retried in the publisher. A constraint violation marks the transaction rollback-only, and that transaction belongs to the caller. Recovering from it would mean destroying the caller’s registration or payment in order to save a signal, which is exactly backwards.

A single atomic statement has no race to lose, no exception to catch, and never puts the caller’s transaction at risk.

The statement uses MySQL’s row-alias form (AS incoming …​ ON DUPLICATE KEY UPDATE col = incoming.col), not the older VALUES() function, which is deprecated from MySQL 8.0.20. Every assignment reads from the alias and never from a column being assigned in the same clause — MySQL evaluates those left to right and later expressions see already-updated values, so a clause that reads its own columns is order-dependent and easy to get subtly wrong.

5.2. Non-squashable signal types

A signal type configured with a zero window is one where every occurrence matters individually — one payment, one email. For those, SignalPublisher stores no aggregate key at all.

That is not an omission. The aggregate key is the squash key: storing one would let two such signals collapse into a single row and silently lose a customer communication. With aggregate_key null the generated pending_squash_key is also null, nothing conflicts, and every emit inserts. Find these signals by entity_type + entity_id instead.

A signal type with no configured window falls back to a non-zero default rather than to zero, so a forgotten configuration entry degrades to "batched more than intended" rather than silently disabling squashing.

Worked example — bulk result import

Fifty race results are imported for series 8. Each emits RESULT_SET_CHANGED with aggregate_key = 'series:8'. The configured window is 60 seconds.

Time Action Table state

T+0:00

Result 1 imported

INSERT, process_after = T+1:00

T+0:01 – T+0:30

Results 2–50 imported

49 squashes; process_after slides to T+1:30

T+1:30

60 seconds of quiet

Processor picks it up — one recalculation covering all 50 results

A window of 0s disables squashing: every emit inserts a new row and is processed on the next poll, and no aggregate key is stored (see Non-squashable signal types). Use this where each occurrence is individually meaningful, such as one communication per order.

Because the payload is overwritten on each squash, the reactor sees only the final state. That is normally what you want — the reactor reads the entity as it now stands. If a reactor needs to know about every intermediate change, squashing is the wrong model for it and its signal type should use a zero window.

6. Framework Components

signal-sweep-architecture

6.1. SignalPublisher

The only part of the framework domain code touches. Fire-and-forget.

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

Transaction participation is deliberate. emit() joins the caller’s transaction. If the caller rolls back, the signal is never persisted — correct, because there is no domain change to react to. The cost is that the signal is not visible until the caller commits; that latency is accepted in exchange for never having a signal that refers to a change which did not happen.

The corollary matters when writing emit calls: emit after the domain change, inside the same transaction. Do not emit from a @TransactionalEventListener(AFTER_COMMIT) or a new transaction — that reintroduces the orphan-signal problem the design avoids.

6.2. SignalProcessor

A single scheduled poller.

@Scheduled(fixedRateString = "${ems.signal-sweep.signal-processor.poll-interval}")
@Transactional
public void processPendingSignals() {
    if (!leaseManager.isSignalMaster()) return;
    if (!runtimeControl.isProcessorEnabled()) return;

    List<DomainSignal> ready = signalRepository.findReadySignals(Instant.now(), batchSize);
    // WHERE status = 'PENDING' AND process_after <= :now ORDER BY process_after ASC

    for (DomainSignal signal : ready) {
        // mark PROCESSING, then for each reactor registered for this signal type:
        //   skip if its signal_reactor_status is already COMPLETED
        //   invoke reactor.handleSignal(signal)
        //   record COMPLETED, or FAILED with backoff
        // finally derive the signal's overall status
    }
}

Key behaviours:

  • Only matured signals are selected. Debounce is already done.

  • A reactor is filtered out for signals outside its organisation set before it is invoked.

  • Reactors are invoked independently; one throwing does not prevent the others from running.

  • A reactor already COMPLETED for this signal is skipped on retry.

  • Failures are recorded per reactor with retry_count and next_retry_at; exhausted retries become DEAD.

6.3. SweepScheduler

Registers each reactor’s sweep against its declared cron at startup.

@PostConstruct
public void scheduleReactorSweeps() {
    for (Reactor reactor : reactors) {
        if (reactor.getSweepCron() != null) {
            taskScheduler.schedule(() -> executeSweep(reactor),
                new CronTrigger(reactor.getSweepCron()));
        }
    }
}

executeSweep guards on isSweepMaster(), the reactor’s effective enabled state, and — for an organisation-scoped reactor — a non-empty effective organisation set, then calls reactor.sweep() inside a try/catch. A failing sweep is logged and does not unregister the schedule.

A scoped sweep must apply the organisation set to its own selection query. The framework cannot verify that, so it enforces what it can: it refuses to start the sweep at all when the set is empty.

Sweeps must be idempotent. They will run again on the next tick, on another pod after failover, and on demand from the admin API.

6.4. The Reactor interface

public interface Reactor {

    /** Unique name — used in config keys, logging, metrics and the admin API. */
    String getName();

    /** Signal types this reactor handles. Empty set = sweep-only. */
    Set<String> getHandledSignalTypes();

    /** Handle one signal. Throwing marks this reactor FAILED and schedules a retry. */
    void handleSignal(DomainSignal signal);

    /** Periodic reconciliation over domain state. Must be idempotent. */
    void sweep();

    /** Spring cron expression for the sweep. Null = signal-only. */
    String getSweepCron();

    /** Whether this reactor acts on tenant data. Defaults to true. */
    default boolean isOrganisationScoped() { return true; }
}

Reactors are discovered as Spring beans. Implementing the interface and annotating the class is all that is needed to register one — there is no central registry to edit.

Note what the interface does not contain: there is no coalesce or debounce method. Debounce is a framework concern configured per signal type, not a reactor decision, because one signal type may feed several reactors that all benefit from the same window.

6.5. LeaseManager

See Multi-Node Leader Election.

6.6. RuntimeControlService

See Runtime Control.

7. Multi-Node Leader Election

admin-service runs on several Kubernetes pods. Without coordination every pod would poll signal_domain and run every sweep, duplicating work and contending on the same rows.

Two independent leases are elected, held in signal_sweep_lease:

Role Responsibility

SIGNAL_MASTER

The one pod that runs SignalProcessor. Every pod still emits signals — emission is just an insert — but only this pod processes them.

SWEEP_MASTER

The one pod that executes reactor sweeps.

Separating the roles allows them to land on different pods, though in practice one pod usually holds both. On a single-pod deployment that pod holds both automatically.

7.1. Lease protocol

  1. On startup each pod generates a unique holder_id — ${HOSTNAME} plus its startup timestamp, so a restarted pod on the same hostname does not inherit its own stale lease.

  2. Every heartbeat interval, each pod attempts to acquire or renew each lease with a single INSERT …​ ON DUPLICATE KEY UPDATE whose conditional logic renews only its own lease and claims another’s only once expired.

  3. The pod reads the row back. If holder_id matches, it is the leader.

  4. SignalProcessor and SweepScheduler each begin with a one-line guard.

The TTL must exceed the heartbeat interval or leases will flap. With the defaults (heartbeat 15s, TTL 30s), a crashed leader’s work resumes within roughly TTL + heartbeat = 45 seconds. That gap is harmless: signals are durable and simply wait.

7.2. Why not SELECT …​ FOR UPDATE SKIP LOCKED

Distributing signal rows across pods by row-locking was considered and rejected:

  • Squashing assumes a single writer’s view of what is pending; splitting processing across pods undermines coalescing.

  • Sweeps scan domain tables, not the signal table — they cannot be partitioned this way.

  • One master means one place to look for logs and metrics.

At the expected volume — hundreds to low thousands of signals a day — throughput is not the constraint. If it ever becomes one, the framework can shard by signal type without changing the reactor contract.

8. Runtime Control

Reactors, the processor and the scheduler can each be enabled or disabled at runtime, in the same spirit as changing a log level in production. There are three levels:

Level Behaviour

Config

From application.yml, per Spring profile. Survives restarts.

Runtime override

Set via the admin API. In-memory (ConcurrentHashMap), takes precedence while the pod runs, and is deliberately lost on restart — a safety net so an emergency override cannot silently become permanent configuration.

Effective

runtime override ?: config default. This is what the framework checks.

The override is per pod, not cluster-wide. Disabling a reactor during an incident means calling the endpoint on the pod holding the relevant lease, or on all pods.

This matters most in two situations:

  • Development. A developer’s machine should not run sweeps against a shared database or send email. The dev profile turns everything off and the developer enables just the reactor under test.

  • Production incident. A reactor hammering an external API can be stopped immediately, without a redeploy.

9. Organisation Scoping

A reactor that acts on tenant data may only act for organisations it has been explicitly given. The set is framework-level, not per reactor, so a new sweep-based reactor inherits the constraint rather than rediscovering it.

This is a correctness precondition, not a rollout convenience. sales_order is a single table shared by both production lines while only one of them runs this framework, so an unscoped reactor reaches orders belonging to tenants this instance does not serve — minting external orders against the other line’s sites, or emailing its customers. The one-tenant-per-instance operating model depends on the filter.

Empty means none, not all. The configured default is an empty set, so a reactor deployed to an environment where nobody named its tenants processes no signals and runs no sweep. This mirrors enabled defaulting to false: silence must never enable behaviour.

Three rules follow from that:

  • Signal routing skips a signal whose organisation is outside the set. The reactor is filtered out before it is invoked, so the signal never reaches it. A signal no reactor is scoped for completes rather than lingering — nothing is waiting on it.

  • A signal carrying no organisation at all is skipped by an organisation-scoped reactor. An unattributable signal cannot be checked against the set, and guessing is how one line’s reactor ends up acting on the other’s data.

  • SweepScheduler refuses to start a scoped sweep whose set is empty. The framework cannot inspect a reactor’s own query, so it cannot prove the sweep honours the set; refusing to run it at all covers the case that matters.

A reactor declares whether it is subject to the filter through Reactor.isOrganisationScoped(), which defaults to true — the safe answer is the one a new reactor gets by writing nothing. SignalRetentionReactor overrides it to false: it maintains the framework’s own tables, has no tenant to be checked against, and filtering it would disable it permanently.

The set has the same three tiers as enabled and dry-run (see Runtime Control), and for the same reason: narrowing or widening it during a rollout must not require a redeploy. Because the configured default is empty, an override lost to a restart falls back to no organisations rather than all of them — losing the gate fails closed.

The setter is refused with 409 unless the pod holds SIGNAL_MASTER. Reactor flags are read only inside SignalProcessor.poll() and SweepScheduler, both of which return early on a non-master pod, so an override accepted elsewhere would change nothing while reporting success — and for a tenant gate that silent failure is in the widening direction.

Per-environment values belong in the deployment manifest, not in a profile file: application-prod.yml reaches stage and both production lines alike, so it cannot express a per-environment tenant set.

10. Configuration

All framework and reactor settings live under ems.signal-sweep:

ems:
  signal-sweep:
    enabled: true                        # master kill switch
    lease:
      ttl: 30s                           # must be greater than heartbeat
      heartbeat: 15s
    signal-processor:
      poll-interval: 5s
      batch-size: 50
    retention:
      days: 7                            # purge COMPLETED/DEAD after this
    debounce:                            # per signal type
      RESULT_SET_CHANGED: 60s
      REGISTRATION_COMPLETED: 0s         # immediate — one signal, one action
      PAYMENT_COMPLETED: 0s
      PARTICIPANT_CATEGORY_CHANGED: 15m
    reactors:
      <reactor-name>:
        enabled: true
        sweep-cron: "0 */10 * * * *"
        organisations: []                # tenants this reactor may act for; empty means NONE

Debounce is keyed by signal type, not reactor, because a type consumed by several reactors should mature once for all of them.

Each reactor owns the ems.signal-sweep.reactors.<name> subtree and may add its own keys there — a dry-run flag, thresholds, age limits. Keep reactor-specific settings inside that subtree rather than inventing a parallel namespace.

11. Retention

COMPLETED and DEAD signals are purged after the retention window (default 7 days) by a framework-internal sweep-only reactor. signal_reactor_status rows cascade with their parent.

Short retention is intentional: signal_domain is a work queue, not an audit log. The durable record of what happened lives in the domain entities and, for communications, in communication_log. Signals in PENDING, PROCESSING or FAILED are never purged regardless of age — an old row in one of those states is a fault worth seeing, not clutter.

12. Monitoring API

SignalAdminResource at /api/admin/signals, ROLE_ADMIN only.

Every endpoint returns 403 when the caller lacks ROLE_ADMIN, so it is omitted from the table below. The codes shown are the ones the published OpenAPI spec carries — springdoc reads annotations only, so an endpoint’s javadoc reaches maintainers but never a generated client.

Note the toggles return 204, not 200: they carry no body.

Endpoint Purpose Responses

GET /signals

Paginated list, filterable by status, signal type, aggregate key and date range. Filters resolve strictly — an unknown value is an error, not silently unfiltered rows

200 400

GET /signals/{id}

One signal including its per-reactor delivery statuses — this is how you find which reactor failed

200 404

GET /signals/stats

Counts by status and type, oldest pending

200

GET /signals/stuck

Signals in PROCESSING beyond a threshold — a hung reactor or a processor that died mid-flight

200 400

POST /signals/{id}/retry

Reset a DEAD signal to PENDING. 409 means a fresher PENDING signal already holds the same (signalType, aggregateKey) — the body names it as supersededByPendingId, the work is already queued, and the DEAD row is left untouched

200 400 409

POST /signals/{id}/cancel

Mark a PENDING signal COMPLETED without processing it

200 400

GET /signals/reactors

All registered reactors: handled types, sweep cron, last sweep, and the config/runtime/effective state of enabled, dry-run and the organisation set

200

PUT /signals/reactors/{name}/enabled

Runtime toggle for one reactor

204 400 404

PUT /signals/reactors/{name}/dry-run

Runtime dry-run toggle for one reactor — 409 unless this pod holds SIGNAL_MASTER

204 400 404 409

PUT /signals/reactors/{name}/organisations

Runtime override of the tenants one reactor may act for — 409 unless this pod holds SIGNAL_MASTER. An empty list pins the reactor to no organisation; see Organisation Scoping

204 400 404 409

POST /signals/reactors/{name}/sweep

Trigger a sweep immediately, bypassing the schedule but not the Sweep Master lease — 409 if this pod does not hold it

200 404 409

PUT /signals/processor/enabled, PUT /signals/sweeps/enabled

Framework-level toggles

204 400

GET /signals/status

Whole-framework overview

200

GET /signals/leaders

Which pod currently holds each lease — start here to find the pod a 409 should be reissued against

200

A 409 on any of these is a routing answer, not a failure: the operation is valid but this pod is not the one that can perform it. GET /signals/leaders names the pod that can.

Diagnostic starting points:

  • Nothing is being processed — check GET /signals/leaders (is anyone holding SIGNAL_MASTER?) then GET /signals/status (is the processor enabled?).

  • One reactor is enabled but does nothing — check its organisation set in GET /signals/reactors. An organisation-scoped reactor with an empty effective set is inert by design; see Organisation Scoping.

  • One concern stopped working — GET /signals?status=FAILED and read error_message, or GET /signals/{id} for the per-reactor breakdown.

  • Work is slow to happen — check the debounce window for that signal type; a long window plus continuous activity means process_after keeps sliding.

13. Observability

Beyond the API:

  • Logging — structured entries for emission, squash, processing, reactor failure and sweep execution, each carrying the reactor name and signal id.

  • Metrics (Micrometer) — counters for signals emitted, processed and failed per type; sweep duration histograms per reactor.

14. Related Documentation