Documentation / Exploitation / Guides pratiques / Connect an event source and activate bindings

Connect an event source and activate bindings

This recipe wires up event-driven ingestion on a running instance: declare an inbound connection whose type is a source mechanism (e.g. fhir-poll), give it its parameters, and choose which of the bindings contributed by your plugins it enables. As an operator you own the connection: an entry of connections: in the instance manifest (libs/shared/src/plugin/connections.ts), or a connection created in the console. The query each binding runs and the map it uses to project a raw record into a domain event are authored by the plugin, not configured here — see Contribute an event source and binding and The open event model for that side.

An event source is an inbound connection type, like http, directory and mllp: its connections are listed in Connexions → Connexions entrantes, and the records the source reads go through the reception inbox before their binding projects them (see Reception of a record).

Before you start

  • The plugin contributing the source mechanism and the bindings you want must already be installed — see Declare and install plugins. Sources and bindings are plugin contributes.eventSources / contributes.eventBindings entries (libs/shared/src/plugin/manifest.ts); a connection can only enable what a loaded plugin contributes, never invent one.

  • You need the exact name of the source mechanism and of each binding on it. List what's loaded:

    curl -s http://localhost:4000/api/plugins | jq '.[].contributes.eventBindings'

    The formats a connection can receive list each source with the bindings a connection of its type can enable (pulled: true; event is the event type, or resolve):

    curl -s http://localhost:4000/api/connections/protocols | jq '.items[] | select(.pulled) | {protocol, bindings: [.bindings[] | {name, event}]}'

    You can also read the plugin's plugin.yaml directly, or open the plugin on the console's Plugins page, whose Events tab lists each binding with its source, query and event (see Inspect loaded plugins). For example, external-plugins/fhir/plugin.yaml contributes one mechanism, fhir-poll (external-plugins/fhir/sources/poll.ts), and several bindings on top of it — fhir-creatinine, fhir-medication-prescribed, fhir-patient-admitted, fhir-patient-discharged, fhir-allergy-recorded, fhir-communication-received, fhir-condition-recorded, plus one resolve-style binding, fhir-validation-decided, that resumes suspended runs instead of starting new ones.

  • To see what a binding produces, open Plugins → Events, section Declared events (« Événements déclarés »): each declared event lists its contract and, in the column Produced by (« Produit par »), the bindings that produce it. The same contracts are served by the catalogue:

    curl -s http://localhost:4000/api/catalog | jq '.eventTypes'

  • A source that reads an external system through a connection (fhir-poll reads a fhir-r4 connection) needs that outbound connection declared, or a default connection of its type: see Manage connections.

1. Declare the connection

In the instance manifest

Add an entry under connections: whose type is the mechanism's name (not a binding's name):

connections:
  entrepot:
    type: fhir-r4
    baseUrl: https://fhir.chu.example/r4
  fhir-poll:
    type: fhir-poll
    label: Interrogation de l'entrepôt
    connection: entrepot
    intervalSeconds: 15
    bindings: [fhir-creatinine, fhir-validation-decided]
  • The connection has no format: its format is the source itself.
  • bindings lists the bindings of the source to enable (step 2).
  • enabled (default true) set to false stops the source and holds its pending records.
  • auth stays none (the default): a pulled source verifies no caller. Another scheme makes the connection unusable, with the issue auth « <scheme> » : une source tirée ne reçoit pas d'appel à vérifier.
  • The other keys are the source's parameters. When the source declares them (params in its plugin manifest), they are checked like the parameters of a connection type: a key the source does not declare, a missing required parameter, or a value of the wrong type or out of bounds makes the connection unusable (paramètre « <name> » …), and the declared defaults apply. A source that declares no parameters receives them unchecked.
  • For a source that requires a connection, the parameter connection names the outbound connection it reads; without it, the source uses the default connection of the type.

For the bundled fhir-poll mechanism, the parameters are connection (the connection of type fhir-r4 to poll, default: the default connection of the type), intervalSeconds (poll period, a number ≥ 1, default 15), count (page size, an integer ≥ 1, default 50), maxPages (pages read at most per feed and poll, an integer ≥ 1, default 10, see Pages and cursor of a feed), startOverlapSeconds (window re-read at startup, a number ≥ 0, default 600, see Restarts and the overlap window) and requestTimeoutSeconds (time limit of each request to the server, a number ≥ 1, default 30). The full list is in Instance manifest.

The identifier of the connection (fhir-poll above) prefixes the id of each event it produces (<connection id>:<feed>:<recordId>). Keep it stable: under another identifier, the same records are new messages and new events.

A connection of the manifest is read-only in the console.

In the console

Connexions → Connexions entrantes → Nouvelle connexion:

  1. Choose the Format named after the source (« Interrogation FHIR » for fhir-poll). Its only transport is the source.
  2. Enter a name. The form shows the parameters the source declares, with their defaults. The Connexion parameter lists the outbound connections of the type the source requires; Connexion par défaut du type leaves it empty.
  3. Under Événements canoniques, tick the bindings to enable.
  4. Tick Activer la réception and save.

A connection created in the console has a generated identifier, which prefixes its event ids. A change to it applies without a restart: the reception stops the source and starts it again with the new parameters and bindings.

2. Choose which bindings are enabled

Each binding enabled on the connection is one feed of the source: the source runs the binding's query, and the binding projects the records of that feed. Name them under bindings (manifest) or tick them (console):

    bindings: [fhir-creatinine, fhir-medication-prescribed]

Without bindings, a connection of the manifest enables every binding of its source. A name that is not a binding of the source, or a binding the host dropped at load, makes the connection unusable: Binding inconnu pour ce protocole : <name>. Two connections of the same source each run their own polls, with their own parameters and bindings.

3. Restart the API

The instance manifest is read once at boot (apps/api/src/instance.ts, loadInstance — the INSTANCE_CONFIG env var if set, else instance.yaml at the repo root). Changes to a connection of the manifest are not hot-reloaded — restart the API process to pick them up. A connection created in the console needs no restart.

The reception runs on one proxy replica at a time: the replica that holds the reception lock (a Postgres advisory lock on the application database, shared with the other inbound connections) starts the sources and delivers their records. The other replicas poll nothing and take over when that replica stops.

Reception of a record

Each record the source emits (feed, recordId, data) is stored in the reception inbox (the inbound_messages table of the application database, DEFINITION_STORE_URL) as the JSON {feed, recordId, data}, with the message id <feed>:<recordId>. The source's emit resolves once the record is stored. A record the connection already received (same feed, same recordId) is a duplicate, even when its content changed: it is not stored or delivered again.

Delivery then projects the record with the binding of its feed:

  • a map binding builds the event, with the id <connection id>:<feed>:<recordId>, and starts the runs of every published workflow that receives its type;
  • a resolve binding resumes the waiting validation it names.

The traffic history of the connection (Historique du trafic, on the Routing screen) lists its records like the messages of the other inbound connections, with the source name as format.

The inbox keeps each record as the source emitted it (a FHIR resource for fhir-poll, hence health data), like HL7 v2 and PN13 messages, and nothing purges it: include inbound_messages in the backup and retention policy of the application database (see Back up and restore).

Records that fail to process

  • When delivery hits an infrastructure failure (Restate ingress unreachable while starting or resuming a run, a run start refused by the ingress), the message is marked delivery-unavailable (error Livraison indisponible. Nouvelle tentative automatique.) and delivered again with the inbox delay: 1 s after the first failure, then 2, 4, 8, 16, 32, and 60 s at most between attempts. The proxy logs livraison indisponible, nouvelle tentative automatique with the source, feed, record id and error, plus eventId and failed (the workflow file and error of each run not submitted) when a submission failed, and counts meridian.events.generated with outcome: "unavailable". The runs already started are kept: the retry starts only the missing ones. Nothing is dropped after a number of attempts.
  • A projection that fails, a record filtered by the binding's require, or an event the ingestion rejects is counted (see Troubleshooting) and its message is delivered: it is not retried.
  • A binding no longer enabled on the connection: its pending records are delivered without projection.
  • A disabled connection holds its pending records until it is enabled again.
  • Réessayer les messages en attente in the traffic history brings the next attempt of the pending records forward.

Restarts and the overlap window

fhir-poll keeps one cursor per feed, in memory (see Pages and cursor of a feed). After each start of the source (a start of the API process, or a change to a console connection), the initial cursor of a feed is the start time minus startOverlapSeconds (default 600):

  • resources modified before the restart and not yet read are read, provided the feed had read its changes up to less than startOverlapSeconds before the restart. Changes older than that window, not read by the stopped process (a stop longer than the window, or a feed behind on its backlog), are not read: raise the value before a planned stop that may exceed it;
  • resources already received before the stop are read again. The connection's inbox already holds them: they are duplicates and are not delivered again. Their events would carry the same eventId (<connection id>:<feed>:<recordId>) anyway, so a workflow that already received one reports a duplicate and starts nothing; a workflow published after their first reception does not receive them either;
  • a feed whose binding sets since: epoch (such as fhir-validation-decided) starts at the epoch instead: after each start it reads every resource of its query again (every decided Task for fhir-validation-decided), maxPages pages per poll; the records already received are duplicates. The feeds are polled one after another, so until that re-read is done each of its polls delays the other feeds by the time of maxPages pages.

Pages and cursor of a feed

Each poll searches the resources of the feed updated since its cursor, oldest first (_lastUpdated=ge<instant>&_sort=_lastUpdated&_count=<count>), and follows the next links of the result, up to maxPages pages per feed (default 10, that is 500 resources with the default count).

The cursor of a feed is the time of its last processed record, plus the records of that second already processed. A poll searches from the beginning of that second and skips those records. Blaze compares _lastUpdated to the second, and a transaction gives several resources the same time: a record written later in the same second is read at the next poll, a record separated from the rest of its transaction by count is read on the next page of the same poll, and only a maxPages cut or an error defers it to the next poll. A record already processed is not emitted again.

  • A poll that reaches maxPages logs <feed>: limite de <maxPages> pages atteinte, suite au prochain passage (curseur <instant>), and the next poll starts right away, without waiting intervalSeconds, from the page where the previous one stopped. A large backlog (the overlap window after a restart, the re-read of a since: epoch feed) is read in slices of maxPages pages, the feeds taking turns; while the process runs, nothing is skipped.
  • The next link carries the base URL the server advertises (Blaze: its BASE_URL, http://blaze:8080/fhir in the local stack), which may differ from the base URL of the connection. The source learns that base from the result (self link, fullUrl of the entries) and requests the page on the connection's base URL with the part of the link that follows it, query included (/Observation/__page/<token>, ?_getpages=<token>); otherwise it keeps the path from the resource type, or the path after the one of the base URL. A link that maps onto none of them is not followed: the source logs <feed>: lien de page suivante hors de la baseUrl (<origin>), suite au prochain passage (curseur <instant>), and the next poll searches again from the cursor, one page at a time. When that first page holds only records already processed, the feed does not advance, and the source logs <feed>: lien de page suivante hors de la baseUrl (<origin>) et page déjà traitée, le feed n'avance pas (curseur <instant>). Corriger la base annoncée par le serveur ou relever count.
  • A poll that emitted records logs <feed>: <n> <resource>(s), followed by sur <k> pages when it read several pages.
  • When nothing has changed, each poll reads again the records of the second of the feed's last record, usually one page. Once that second is more than 60 seconds old by the server's clock (the Date header of its response, else the local clock), a search read to the end moves the cursor to the next second and the second is no longer read.
  • A lastUpdated older than the previous one in the result (for instance a server clock that stepped back: Blaze dates each transaction with its wall clock) is emitted like the others; the cursor moves back to the oldest time read since, so that the writes dated by the stepped-back clock are read, and the source logs <feed>: lastUpdated non croissant dans la réponse, curseur ramené à <instant>. Vérifier l'horloge du serveur FHIR. Records already processed are not emitted again.

The cursor moves with each record stored in the inbox. Processing happens at delivery, so a record whose processing fails does not stop the feed (see Records that fail to process). An emission fails only when the record cannot be stored (application database unreachable, connection disabled meanwhile); the poll of the feed then stops, the source logs <feed> échoué, reprise au prochain passage (curseur <instant>) : <error>, and the next poll starts again from that record. A page the server does not return (server unreachable, HTTP error, no answer within requestTimeoutSeconds: délai de <n> s dépassé) stops the poll the same way, and the other feeds are polled.

For the resolve binding fhir-validation-decided, the resolution answers 409 without applying anything when the run is already terminal in the run index, when the Restate ingress answers an HTTP error for the run (a run past the Restate retention window), when the run no longer awaits a validation, when the decision names another validation of the run, or when Restate refuses the decision because the validation was already resolved. The record is counted as a duplicate and its message is delivered. When the ingress is unreachable, the delivery is retried and the decision is applied once Restate answers again.

The binding correlates a decision with one validation, not only with its run. The Task carries two identifiers, both written when the platform publishes the validation: urn:ehr-lab:validation-run (the run) and urn:ehr-lab:validation-pending (the validation, pending.id of the run). A run can wait on several validations in turn, and each startup re-reads every decided Task from the epoch: a decision is applied only if its validation is the one the run is waiting on. A Task without the urn:ehr-lab:validation-pending identifier is rejected (résolution rejetée, validation visée absente de l'enregistrement : décision non appliquée).

An HTTP error from the ingress is not always a run past retention. A 5xx answer is treated as a passing failure: the proxy logs résolution indisponible, nouvelle tentative automatique with the run id and the error, and the reception retries the delivery of the record until it passes. Another HTTP error counts as a 409 and the decision is not applied: the proxy logs résolution non appliquée with the run id and the ingress status, the record is delivered, and the run stays suspended until someone resolves it from the console or the Task is written again (a new version is a new record).

Meridian's own writes come back as events

Records written by workflows are read back by the feeds like any other change: a note written to the record (notify.send with toRecord) comes back through fhir-communication-received as a CommunicationReceived event with the sender « meridian », and a condition added to the record comes back through fhir-condition-recorded as ConditionRecorded. Every published workflow of that type receives it. Workflows on CommunicationReceived exclude these notes with the trigger filter event.sender != "meridian". That expression is the default filter (defaultSelector) that @posos/common declares on the event CommunicationReceived; it reads the event's sender field. The console prefills it when an author picks a trigger of that event. ConditionRecorded has no field that marks Meridian's own writes. See Choose which events start a workflow and Events and triggers.

Verify

  • GET /api/connections lists the connection with direction: inbound, the source name as protocol and transport, its params and bindings, and status: listening once its source started; the console shows it in Connexions entrantes with the status En écoute.
  • The source's own log lines go to the sources component, with the field connection (the connection's identifier): fhir-poll logs polling … — feeds: <feeds> at start, with the enabled bindings as feeds, then one line per poll that emitted records.
  • GET /api/connections/<id>/messages lists the last records received by the connection, with their status (pending, delivered), attempts and error; the traffic history shows the same. No record stays pending once every record has been delivered.
  • Confirm real traffic turns into activity: watch GET /api/runs for new runs after data lands upstream, or check the meridian.events.generated counter (attributes source, the source name, feed, event, outcome) for outcome: "accepted" — see Enable observability.
  • A binding that produces a declared event is checked against the event's contract when the plugins load. A mismatch logs « binding hors contrat » (component plugins) and the binding stays active; see Observability signals.
  • For a resolve binding — it resumes a suspended validation instead of starting a run, see Add human validation — confirm the targeted run leaves its pending state once the upstream decision lands.

Troubleshooting

  • connexion entrante « <id> » du manifeste inutilisable : … at boot, and the connection shown with the status error: the issues name the cause, such as a parameter the source does not declare or of the wrong type (paramètre « <name> » …; a numeric string such as "15" is not a number), a binding the source does not have (Binding inconnu pour ce protocole : <name>.), or an auth scheme other than none. The source does not start.

  • connexion « <id> » du manifeste inutilisable : type « <source> » non contribué par les plugins chargés.: no loaded plugin contributes the source, so the entry is read as an outbound connection of an unknown type. Confirm the plugin is installed and the exact source name via GET /api/plugins.

  • Source non démarrée. Vérifiez ses paramètres et sa connexion. (status error): the source's start() failed. Connexion non démarrée. Vérifiez ses paramètres et les droits du process.: its factory threw.

  • <feed> échoué : Aucune connexion par défaut pour le type « fhir-r4 ». … or <feed> échoué : Connexion « <id> » introuvable.: the outbound connection of the source does not resolve. Set the connection parameter, or a default connection of the type (see Manage connections).

  • binding ignoré : événement « <type> » non déclaré par les plugins chargés. Charger le plugin qui le déclare. or binding ignoré : « <type> » est l'adresse du déclencheur épinglé « <trigger> ». Viser un événement déclaré. (component plugins, fields plugin, binding, source, event): no loaded plugin declares the binding's event, or its type is the address of a trigger with its own address, such as the reception of a raw message. The host drops the binding at load: a connection that names it is unusable, and a connection without bindings leaves it out. The Plugins page marks it ignored at load. Load the plugin whose trigger receives the event, or ask the plugin author to fix the binding's event.

  • <feed>: lien de page suivante hors de la baseUrl (<origin>): the next link of the server maps onto neither the base it advertises nor the connection's base URL, and the feed advances one page per poll, as long as each first page holds a new record. With … et page déjà traitée, le feed n'avance pas, the first page holds only records already processed and the feed stays in place: set the server's advertised base (Blaze: BASE_URL) to the address the source reaches, or raise count above the number of records of the feed that share one second.

  • <feed>: lastUpdated non croissant dans la réponse, curseur ramené à <instant>: the server's clock stepped back, or the server does not sort by _lastUpdated. Nothing is skipped; records already processed are not emitted again. Check the clock of the FHIR server.

  • livraison indisponible, nouvelle tentative automatique: delivering the record threw; the error is in the log line. The record is delivered again until it passes (see Records that fail to process). When Restate is down, the records stay pending and the backlog drains once it answers again.

  • A validation decision is not applied: the Task decides another validation of the run than the one the run waits on (409, counted as a duplicate), or it lacks the urn:ehr-lab:validation-pending identifier (résolution rejetée). The system that decides must update the Task still requested for the run, not an older one.

  • Source is polling but no runs appear: check the meridian.events.generated counter by outcome and the sources log.

    • The binding's map fails: projection (map) échouée, counted rejected.
    • A field listed in the binding's require is absent (missing, null, blank text, or [] for a list field of the contract) or, under EVENT_CONTRACTS=enforce, a field or identifier the contract requires is: the record is not emitted and is counted rejected, with the debug log enregistrement filtré (champs exigés vides).
    • No published workflow receives the type: Aucun workflow pour l'événement <type>, counted rejected.
    • Under EVENT_CONTRACTS=enforce, the event does not conform to its contract: Événement « <type> » non conforme à son contrat., counted rejected; the Routing traffic history lists its issues (see Contract check).
    • Every receiving workflow has a filter and none keeps the event: counted filtered, shown Filtered (« Filtré ») in the Routing traffic history.
    • A workflow's filter fails on the event: that workflow does not start and the API logs Filtre en erreur — « <file> » ne reçoit pas l'événement <type>. (Filter in error / « Filtre en erreur » in the traffic history).
    • A resource updated upstream under the same recordId (a fhir-poll binding with dedupe: id) is a duplicate for the connection and starts nothing, see Reception of a record.

    The binding, the filters and the workflows are plugin and workflow authoring concerns, not instance configuration — see Contribute an event source and binding and Choose which events start a workflow.

97 documents14 sectionssource : /docs · généré au build