Alerting Implementation

The alerting mechanism (the RULE directive) is an integral part of the main processing loop. It is not a separate background process - rules are evaluated synchronously, in the same time-grid iteration as the SELECT computations. This guarantees that an alert always refers to data that was just computed, not to the previous cycle.

Where RULE sits in the processing cycle

Recall the processRows() function outlined in the chapter Query Tree Traversal Algorithm. For every non-declaration query, four steps are carried out in sequence (Fig. 49):

%%{init: {"markdownAutoWrap": false}}%%
flowchart LR
    A["constructInputPayload()"] --> B["constructOutputPayload()"]
    B --> C["write()"]
    C --> D["constructRulesAndUpdate()"]

Fig. 49. The order of processing steps for a single query

The fourth step - constructRulesAndUpdate() - is exactly where all rules attached to the current query are executed. It is called after the SELECT results have been written to disk, which means a rule always evaluates against a complete, just-computed sample of the stream.

Evaluating the WHEN condition

Every rule contains a list of tokens describing a logical expression (the condition field of the rule struct). At the moment of evaluation, the system:

1. Fetches the current query’s outputPayload - the current sample of the stream.
2. Passes the condition to the expressionEvaluator::eval() engine - the same engine that computes SELECT expressions.
3. Casts the result to a boolean (boolCast): any non-zero numeric value is true, zero is false.

If the condition is satisfied, the action associated with the rule is executed (DO SYSTEM, DO DUMP, or DO SET). If not, the rule is skipped with no side effects at all. The full flow is shown in Fig. 50.

%%{init: {"markdownAutoWrap": false}}%%
flowchart TD
    A["New stream sample"] --> B["expressionEvaluator::eval(condition, sample)"]
    B --> C{boolCast}
    C -->|true| D{action type?}
    C -->|false| E([skip])
    D -->|DO SYSTEM| F["system(command)"]
    D -->|DO DUMP| G["dumpManager::registerTask()"]
    D -->|DO SET| I["Evaluate value and buffer update"]
    I --> H
    F --> H["dumpManager::<br/>processStreamChunk()"]
    G --> H

Fig. 50. Rule evaluation flow

The DO SET action and the slot boundary

constructRulesAndUpdate() evaluates the SET right-hand side on the same record and VAR values as WHEN. It buffers the result, including NULL, for the current slot. Competing writes retain the update from the later rule in plan text order. Ad hoc rules receive subsequent positions.

Only successful completion of dataModel::processRows, after advancing declared sources, copies the buffer to current values. Every field and rule in the slot reads the previous state; node order does not change update visibility. A failure before this boundary commits no VAR updates, and the next slot clears the buffer; this does not roll back stored records or external actions. The same boundary applies in --until-eof and the embedded engine. xqry --vars reads between slots under the plan epoch lock.

Loading or resetting a plan constructs values from declarations, so rebuilding a window alone cannot restore a latch lost on restart. See VAR and DO SET for semantics and restrictions. Coverage: ut_dataModel, it_var_set-value, it_var_reset-run.

The DO SYSTEM action

The DO SYSTEM invocation is the simplest: the system calls ::system(command) directly on the processing thread. The call is synchronous - xretractor waits for the process to finish before moving on to the next rule.

The command’s exit code is checked:

• 0 - success, no log entry.
• ≠ 0 - xretractor logs an error via spdlog with the exit code.
• A system() failure (e.g. no shell available) - logged as a critical error.

⚠️ Warning

The command is executed synchronously. Long-running scripts (e.g. sending large files, network calls with a timeout) will delay the entire processing cycle. In such cases, it is recommended to launch the process in the background: DO SYSTEM 'my_script &'.

The DO DUMP action - detailed algorithm

DO DUMP writes the requested range of records relative to sample t, for which the rule condition is true. The range is half-open: [step_back, step_forward), with step_back < step_forward; the record at offset step_forward is excluded. Negative offsets refer to history, 0 to the current sample, and positive offsets to future samples. This is handled by the dumpManager class.

Phase 1: historical data (when the task is registered)

At the moment the rule fires - right after the condition is found to be true - dumpManager::registerTask():

1. Removes any existing entry at the dump filename (unlink()) and creates a new file with POSIX open() using the O_RDWR | O_CREAT | O_EXCL | O_NOFOLLOW | O_CLOEXEC flags. It then removes earlier tasks writing under the same filename from the queue and closes their descriptors (see Retention).
2. If step_back < 0, reads the historical part [step_back, min(step_forward, 0)), which contains min(step_forward, 0) - step_back records. The compiler accounts for the furthest required historical offset when calculating the stream’s capacity.
3. Writes these records to the file from oldest to newest, excluding the record at the right boundary of the historical range.
4. Computes the number of remaining records: dumpedRecordsToGo = (step_forward - step_back) - historyWritten, where historyWritten is the number of records written in step 3, or zero when the range has no historical part.
5. If step_back >= 0 (a delayed start), it sets delayDumpRecordsToGo = step_back.

Example: DUMP -3 TO 2
  At registration: write samples t-3, t-2, t-1 (history)
  Still to collect: 2 samples (t, t+1)
  dumpedRecordsToGo = 2

Example: DUMP -5 TO -2
  At registration: write samples t-5, t-4, t-3
  dumpedRecordsToGo = 0 - the whole range has been written

Example: DUMP -5 TO 0
  At registration: write samples t-5, t-4, t-3, t-2, t-1
  dumpedRecordsToGo = 0 - sample t is outside the range

With step_forward <= 0, the entire dump is ready after task registration. The next processStreamChunk(), called while sample t is still being handled, removes the completed task and closes its file; it appends neither the current sample nor future ones.

For a rule from the plan file, a DUMP -H TO M range (H > 0) requires at least H+1 records: H historical records plus the current record. The compiler automatically enlarges the STORAGE MEMORY ring to this capacity, including when RETENTION is omitted or specifies a smaller value. A larger explicit RETENTION remains the ring’s minimum capacity. For DEFAULT and DIRECT file stores with bounded retention, the compiler rejects the plan if retention cannot provide H+1 records even immediately after rotation: (segments - 1) * capacity + 1 >= H+1 must hold. This also applies to retention from the [storage] default_retention configuration setting.

For a rule attached ad hoc, the entire history must be produced after attachment. With a DUMP -H TO M range, the rule can first evaluate its WHEN condition on record H+1 after attachment: the preceding H records provide the history, and the new record is the current sample. Without a historical part (H=0), the condition is evaluated on the first new record. A MEMORY stream must retain at least H+1 records; a request reaching further back is rejected without attaching the rule.

Phase 2: the current sample and future data

After registration, the task goes into the bookOfTasks[streamName] queue. dumpManager::processStreamChunk() is called after all rules have been evaluated for sample t, and then after every new sample of the stream:

1. For every active task in the queue (dumpedRecordsToGo > 0):
   • If delayDumpRecordsToGo > 0 - decrement and skip (start delay).
   • Otherwise - write the current sample to the file and decrement dumpedRecordsToGo.
2. Once dumpedRecordsToGo reaches 0 - close the file descriptor and remove the task from the queue.

The full sequence for DUMP -3 TO 2 is shown in Fig. 51.

%% pdf-width: 85%
%% pdf-height: 55%
%%{init: {"markdownAutoWrap": false, "sequence": {"mirrorActors": false, "messageMargin": 22, "boxMargin": 6}}}%%
sequenceDiagram
    participant SI as streamInstance
    participant DM as dumpManager

    note over SI: Sample t - condition TRUE
    SI->>DM: registerTask(stream, {-3, 2, retention=0})
    DM->>DM: Open file dump.tmp
    DM->>DM: Write t-3, t-2, t-1 (history)
    DM->>DM: dumpedRecordsToGo = 2
    SI->>DM: processStreamChunk(stream)
    DM->>DM: Write t → dumpedRecordsToGo = 1

    note over SI: Sample t+1
    SI->>DM: processStreamChunk(stream)
    DM->>DM: Write t+1 → dumpedRecordsToGo = 0
    DM->>DM: Close the file - task complete

Fig. 51. Data-collection sequence for DO DUMP –3 TO 2

The delayed-start case (step_back ≥ 0)

When step_back is non-negative, the dump does not start at the moment of the event, but step_back samples after it:

Example: DUMP 2 TO 5
  At registration: delayDumpRecordsToGo = 2
  Sample t   → skip (delay=2→1)
  Sample t+1 → skip (delay=1→0)
  Sample t+2 → write (dumpedRecordsToGo = 3→2)
  Sample t+3 → write (dumpedRecordsToGo = 2→1)
  Sample t+4 → write (dumpedRecordsToGo = 1→0) - done

Retention (RETENTION N)

Without a RETENTION clause, every trigger of a rule writes its dump under a single name <stream>_<rule>_dump.tmp. With a RETENTION N clause, the first trigger of a rule in an engine run creates _dump_0.tmp, and subsequent file numbers rotate modulo N: _dump_0.tmp, _dump_1.tmp, …, _dump_(N-1).tmp.

A filename must identify a single rule. A plan in which two DO DUMP rules share the stem <stream>_<rule> - e.g. rule b_c on stream a and rule c on stream a_b - is rejected on load, and a rule attached ad hoc with such a stem is refused. The comparison ignores letter case, because on a case-insensitive file system (macOS by default) Ab_r and ab_R are the same file, and plan validity must not depend on the host. The plan author then renames the rule or the stream.

A dump file is always created anew. The engine removes whatever sits under its name - including a symbolic or hard link, which it does not follow - and creates a new file exclusively (O_EXCL | O_NOFOLLOW). A write therefore does not reach the target of a link substituted at the final dump filename. This protection does not cover substitution of parent directories during path resolution; it is not an atomic guarantee of containment within the storage directory. A process that held the previous dump open still sees its old contents. Failure to remove, create or seek within a dump file is an I/O error: xretractor stops execution with environment class K4 and code EIO (5 on Linux), naming the file and cause. The service plan file is preserved; see xretractor for the policy.

Dump tasks wait in the bookOfTasks queue, one per stream and shared by all of its DO DUMP rules. The queue has no capacity of its own and evicts no tasks: a task ends after collecting its whole range, or when the same rule recreates its file. After creating the new file, registerTask() removes earlier tasks writing under the same name from the queue, and the dumpTask destructor closes their descriptors. A replaced task therefore does not write to a file already detached from the directory, and under each name only the newest trigger collects data.

A rule without RETENTION thus has at most one task in progress, and a new trigger interrupts its previous, unfinished dump. A rule with RETENTION N has at most N tasks: an unfinished dump is interrupted only by the trigger that returns to its file after the numbering wraps, i.e. the N-th next one. Tasks of other rules on the same stream stay untouched - rules do not cut each other’s dumps, including two rules without RETENTION. The number of descriptors open on a stream does not exceed the sum of these limits over its rules.

If step_forward > 0, the last required sample is at offset step_forward - 1 relative to the event and must be written before the trigger that reuses the same file (for regularly spaced events: after N intervals, or one without RETENTION). With step_forward <= 0, the entire dump is written at registration.

Dump file format

The file contains raw binary records with no header at all - every record has the size determined by the descriptor (descriptor.getSizeInBytes()). The format is identical to the format used by stream artifacts, which lets you read it with the xtrdb tool after manually specifying the schema:

$ xtrdb
> storage <path>
> open <stream>_<rule>_dump { <type> <field> }
> list
> quit

Complete record writes and I/O errors

Both when copying history and appending current samples, the engine repeats write() until the entire record has been written. A positive result shorter than the record size indicates progress, after which the remaining bytes are written. EINTR allows at most five consecutive retries; writing more bytes resets that counter. The task’s remaining-record count decreases only after a complete record has been written (#479).

A write making no progress (write() returning 0) yields Errc::IO with errno=EIO; another write error preserves its system cause, such as ENOSPC, while exhausting interrupted-write retries preserves EINTR. The diagnostic identifies the path and the number of bytes written against the record size. xretractor ends such a run with environment class K4 and code EIO, preserving the service plan file. A failure after a partial write may leave the beginning of a record in the file; the write is not rolled back. Completing write() does not mean the data has been synchronized to durable storage. The ut_dumpManager tests cover writes in parts with EINTR, no progress, ENOSPC, the retry limit, and seek and history-write errors.

The dump contract: values only, no NULL and no gaps

A dump is a headless block of bytes. dumpManager writes payload->span() directly, record after record, and creates no companion file beside it: there is no .desc, so the schema has to be known from outside, and no .meta, so the NULL map and the transmission gaps have nowhere to go. The .tmp extension suggests a working file, but this is the final artifact - no rename follows once the descriptor is closed.

One consequence follows, and it has to be known before a dump is used as input for anything:

A zero in a dump file is therefore indistinguishable from a genuine zero, from NULL, and from a record the engine never had. This is not an implementation oversight but the boundary of the format: NULL and the gap are notions of the engine’s interior - they live in the payload’s NULL map and in the .meta index accompanying an artifact, they are stored there and processed there - and the system has, as of today, no uniform way of writing them down on the outside. A dump is a snapshot of values, not a record of what the engine knew.

Where fidelity is required, two other routes do carry the information about absence:

• the stream artifact (SELECT … STREAM) together with its .meta file - xtrdb prints the NULL map and the gaps through the meta and metaraw commands (→ Files);
• the client channel xqry --jsonl, where an absent value is a separate JSON null, per element of an array field (→ Stream monitoring API).

The null2zero function in a query is not a third route: it turns absence into a zero explicitly and inside the query, which makes it a lossy conversion rather than a way of exporting the information about absence (→ Aggregate operators).

Multiple rules - evaluation order

Multiple rules can be attached to a single stream. All of them are evaluated in a single constructRulesAndUpdate() iteration, in the order they were declared in the .rql file. Every rule is independent - one being satisfied does not affect the evaluation of the others (Fig. 52).

%%{init: {"markdownAutoWrap": false}}%%
%% pdf-width: 100%
flowchart TD
    A["New sample of stream S"] --> R1["Rule 1: WHEN S[0] > 100"]
    A --> R2["Rule 2: WHEN S[0] < 10"]
    A --> R3["Rule 3: WHEN S[0] > 100"]
    R1 -->|true| A1["DO SYSTEM 'notify-send'"]
    R2 -->|true| A2["DO SYSTEM 'echo alarm'"]
    R3 -->|true| A3["DO DUMP -5 TO 5"]
    R1 -->|false| X1([skip])
    R2 -->|false| X2([skip])
    R3 -->|false| X3([skip])

Fig. 52. Independent evaluation of multiple rules on the same stream

Practical limitations and notes