Data Flow¶

How emails are processed through the SPOT Platform.

Overview¶

SPOT takes each message raw, parses it once and stores it, then analyses it asynchronously: RabbitMQ carries the email's id, never its content, from intake to the analyzer-orchestrator, which runs a workflow and decides the verdict.

Email Analysis Flow¶

1. Intake¶

Client -> API Gateway (parse, store) -> RabbitMQ -> Analyzer Orchestrator

Flow:

An analyst sends POST /api/v1/analyze with the raw message: a message/rfc822 body, or JSON with the message base64-encoded.
The API Gateway refuses a message larger than 25 MiB (413) before reading it, then parses it with spot_sdk.mime in a worker process bounded in time and memory (422 parse_limit past a limit). The view holds the ordered headers (X-SPOT-* removed: only SPOT sets them), the MIME parts, the decoded text and HTML, the URLs, the attachments' metadata and hashes, and the authentication results of the trusted authserv-ids (mail.trusted_authserv_ids).
It stores the raw message (email_raw) and the view (emails.view_encrypted), both encrypted with ENCRYPTION_KEY, in one transaction.
It chooses the workflow: the one the analyst named, else the first route of spot.yaml that matches the email's view (its source api, its recipients, its authenticated sender domain), else the default workflow (Routing). A route to a workflow that is not loaded is skipped.
It publishes {job_id, email_id, workflow_id, priority, route} to the spot.analysis exchange with the email.analyze routing key, records the job as queued in Redis, and returns the job id at once.

A message refused at intake is not stored. Past intake.max_in_flight messages at once the gateway answers 503, to retry later.

Mail Ingest¶

Retriever -> API Gateway (parse, store) -> Mail Orchestrator -> RabbitMQ -> Analyzer Orchestrator

Retrievers send the raw message, base64-encoded, to the API Gateway (POST /internal/ingest, with their plugin token, scope ingest). The gateway parses and stores it as above, with the retriever's id as the email's source and the SMTP envelope when the retriever sends one, and routes it as a submission is routed: the retriever names no workflow. A route's timeout_ms caps the retriever's wait. The gateway logs the retriever, the workflow and the route, never an address. It then hands the email's id, the workflow and the route to the Mail Orchestrator with the internal key. The Mail Orchestrator publishes them to the same spot.analysis exchange with a job id.

Asynchronous ingest is published to the bulk queue (email.analyze, the orchestrator.analysis queue) and returns accepted once the message is published.
Synchronous ingest (wait_for_verdict) is published to a queue of its own (email.analyze.sync, the orchestrator.analysis.sync queue), which the orchestrator consumes with slots of its own (MAX_CONCURRENT_SYNC_JOBS), so a backlog of bulk analyses never holds it up. It waits up to timeout_ms for the analysis job record in Redis and returns analyzed, with the verdict and the score, or timeout. Its message expires after timeout_ms: one not started in time moves to the bulk queue and is analysed later, and never holds up a fresh synchronous ingest. One whose handling keeps failing (Redis down, say) holds its slot through its retries, as in the bulk queue. On timeout the analysis still completes and is stored. A message that cannot be queued is answered 503, an analysis that fails 502, in the gateway's error envelope.
RabbitMQ is required: an unreachable broker rejects the ingest.

2. Job Processing¶

Analyzer Orchestrator -> Analyzers -> PostgreSQL -> spot.results

Flow:

The orchestrator consumes from the orchestrator.analysis and orchestrator.analysis.sync queues. A message that is not an analysis job, or that names no workflow when there is no default, is recorded as a failed job, with the reason; one that names no job is parked at once in orchestrator.analysis.dead-letter (from the sync queue, it first moves to orchestrator.analysis). When the job cannot be recorded (Redis down), the message is delivered again (see Failing messages).
It loads the workflow (the default one when the message names none) and the email's stored view.
It runs the stages, in the order of their depends_on. Each analyzer gets POST /internal/analyze with its own token and an AnalyzeRequest: the job id, the view, the outcomes of the stages that ran before (previous), the stage's retrieval_limits, and the time left (deadline_ms). It answers an AnalyzerVerdict: a verdict, a score, a confidence and the signals behind them. An analyzer may fetch the raw message or one decoded part from the gateway (scope content:read), and query the Knowledge Store.
The workflow's verdict_policy decides the email's verdict, its score and decided_by (see Workflows). The whole run is bounded by the workflow's timeout_ms.
The result is stored in PostgreSQL: the verdict, the score, decided_by and the route in clear, the OrchestrationResult encrypted, and the signals (analysis_signals), their evidence and location encrypted. A failed or cancelled job keeps its route too.
The orchestrator updates the job record in Redis and publishes {job_id, email_id, workflow_id, verdict, score, decided_by} to the spot.results exchange (analysis.complete).

A job whose analyzer with failure strategy fail did not answer, or that cannot run (no view, a condition on a stage that has not run, a stage cycle), fails: its verdict is unknown, decided by failed, with the reason.

3. Responders¶

spot.results -> Analyzer Orchestrator (dispatcher) -> Responders

The dispatcher consumes orchestrator.dispatch, bound to spot.results / analysis.complete. For each result it calls POST /internal/respond on the responders the workflow's on_verdict names for the verdict, each with its own token, at once: the verdict, the score, decided_by, the stored result's signals and the job's mode. Each call is bounded (10 s) and retried three times, then given up and logged. Each responder called is then recorded as an action in PostgreSQL (analysis_actions), applied when the job is in enforce and the responder answered. The message is acked once every responder has succeeded; when one gave up, the message is delivered again, and every responder of the verdict is called again (each acts once per job_id) and recorded in place of its first record, except that a responder that applied stays applied. A responder may call GET /internal/results/{job_id} (scope results) for the whole result.

The job's mode is decided when the job is queued: at intake for an ingest or a submission, at the request for a re-analysis. It is enforce only when its mail connection (the retriever's entry) and its route both are; a user's submission is observe. It travels with the job: in the queue message, the job's records, the stored analysis, the result message and the retriever's IngestionResult. A retriever reports what it did with the message, or would have done in observe, with POST /internal/jobs/{job_id}/action (scope ingest): the gateway checks the job is the retriever's in the mail-orchestrator's record and records the action in the job's mode.

Shadow workflows¶

spot.results -> Analyzer Orchestrator (shadow) -> Analyzers -> PostgreSQL

The shadow consumer reads orchestrator.shadow, bound to spot.results / analysis.complete too, one result at a time (prefetch 1). For a result that did not fail, it takes the shadow entries of spot.yaml, read with the workflows, and runs each shadow workflow whose match holds for the email (production's own workflow aside, and one not loaded skipped) on the stored view, as a run of its own. Where a shadow workflow's verdict class differs from production's, it stores a shadow difference: the verdicts and scores in clear, the shadow workflow's result encrypted, one per job and workflow. A run that fails is logged by its error's type and skipped; a view or a database it cannot reach raises, and the result is delivered again. The queue holds at most 1 000 results, dropping its oldest (RabbitMQ).

Every run that completes is also counted for its day (UTC), in shadow_tallies, differing or not: the rate of a workflow's differences is read from those counts (Trying a workflow in the shadow).

Benchmark runs¶

Gateway -> orchestrator.benchmark -> Analyzer Orchestrator -> Analyzers -> PostgreSQL

A backtest starts at POST /api/v1/benchmarks with a dataset or the stored mail of a period. The gateway records the run (one platform run runs at a time, which a partial unique index of benchmark_runs keeps) and publishes one message per email, a run and an email, to spot.analysis / email.benchmark. The benchmark consumer reads orchestrator.benchmark one message at a time (prefetch 1), on a channel of its own. It reads the run and skips the message of one that has ended; reads the email's stored view, its label, and for received mail production's latest verdict and its job; runs the run's workflow, as it is now, as a run of its own (run="benchmark"); and records the result in the run. The last result finishes the run, which counts its metrics. It stores no job and no analysis, publishes no result, and calls no responder, so neither the dispatcher, the shadow nor the statistics see it.

A failure to read or record raises, and the message is delivered again. On its last delivery, a failure while the email runs or its data is read ends the email as an error result in place of parking the message. A failure to read the run or record the result is outside that: when the database stays unreachable past the redelivery window, the message is parked and the run stays running, which no other backtest can start beside (Messages in a dead-letter queue gives the way out). A message that names no run and email is refused and parked. A cancel ends the run: the messages still queued are skipped.

4. Status Query¶

Client -> API Gateway -> PostgreSQL, Redis

Flow:

Client sends GET /api/v1/analyze/{job_id}.
The API Gateway reads the stored analysis from PostgreSQL.
A job not stored yet is read from its Redis record (queued, running, or finished and not yet stored); records expire after 24 hours.
The answer carries the status, the route that chose the workflow, and, once finished, the verdict, the score, decided_by, the signals and the result.

5. Cancel¶

Client -> API Gateway -> Analyzer Orchestrator -> Redis, PostgreSQL

DELETE /api/v1/analyze/{job_id} answers from PostgreSQL for a job stored already: 409 when it completed or failed, 200 when it was cancelled. Otherwise the gateway calls the orchestrator's POST /jobs/{job_id}/cancel with the internal key:

A job running in the orchestrator has its analysis stopped: the calls to its analyzers are cancelled.
A job not picked up yet is recorded cancelled at once; when its message arrives, the orchestrator drops it without running it. A retriever's ingest has no record until it is picked up: until then, the cancel answers 404.
The job is recorded cancelled in Redis and in PostgreSQL: unknown, score 0, decided by cancelled. Nothing is published on spot.results, so no responder is called.

A cancel the orchestrator could not record answers 503 and stays pending in the orchestrator, which still drops the job when its message arrives; a retry records it. The orchestrator keeps its pending cancels in memory, so a restart loses them; there is one analyzer-orchestrator per installation.

Sequence Diagram¶

Client      API Gateway      RabbitMQ      Orchestrator      Analyzers      PostgreSQL      Responders
  |              |               |               |               |               |               |
  |--raw msg---->|               |               |               |               |               |
  |              |--parse, store raw + view--------------------------------------->|               |
  |              |--publish id-->|               |               |               |               |
  |<--job id-----|               |               |               |               |               |
  |              |               |--consume----->|               |               |               |
  |              |               |               |--load view------------------->|               |
  |              |               |               |--analyze----->|               |               |
  |              |               |               |<--verdict-----|               |               |
  |              |               |               |--store result---------------->|               |
  |              |               |<--publish-----|               |               |               |
  |              |               |--dispatch---->|               |               |               |
  |              |               |               |--respond------------------------------------->|
  |--status----->|               |               |               |               |               |
  |              |--read result------------------------------------------------>|               |
  |<--verdict----|               |               |               |               |               |

Analyzer Communication¶

Each analyzer receives an AnalyzeRequest:

{
  "job_id": "3f0e7c1a-...",
  "email": {
    "id": "9b2d4e6f-...",
    "source": "api",
    "subject": "Verify your account",
    "from": {"display_name": "PayPal", "address": "security@paypaI.com", "domain": "paypai.com"},
    "headers": [{"name": "Received", "value": "..."}],
    "authentication": [{"authserv_id": "mx.example.com", "method": "dmarc", "result": "fail"}],
    "parts": [{"part_id": "1", "content_type": "text/plain", "...": "..."}],
    "visible_text": "Click here to verify...",
    "urls": [{"url": "http://suspicious-site.com", "host": "suspicious-site.com", "found_in": "text"}],
    "...": "..."
  },
  "previous": {},
  "retrieval_limits": null,
  "deadline_ms": 29950
}

Each analyzer answers an AnalyzerVerdict:

{
  "verdict": "suspicious",
  "score": 72,
  "confidence": 80,
  "signals": [
    {
      "name": "auth.dmarc_fail",
      "weight": 30,
      "evidence": "dmarc=fail for paypai.com",
      "location": "header:Authentication-Results"
    }
  ],
  "summary": "Lookalike sender domain failing DMARC",
  "details": {},
  "model_version": null
}

The analyzer's document holds both models in full.

Error Handling¶

Analyzer Failures¶

An analyzer that does not answer an AnalyzerVerdict (a timeout, an HTTP error, an invalid, encoded or oversized answer, no token, no URL) is handled by its failure strategy:

retry: tried again after a backoff, up to retry_config.max_attempts; an analyzer without a token or a URL is not tried again
skip, or retry once its attempts are spent: recorded as unknown, with the reason
fail: the job fails, unknown

An unknown analyzer does not count in the score. When none counts, or a required one is unknown, the email is unknown: a failure is never benign.

Failing messages¶

The queues are quorum queues with a delivery limit of 10. A message whose handler fails is handed back after a wait that starts at 1 s and doubles up to 5 minutes. Once it has failed 10 times, its next failure parks it in its dead-letter queue (orchestrator.analysis.dead-letter, orchestrator.dispatch.dead-letter, orchestrator.shadow.dead-letter, orchestrator.benchmark.dead-letter), about 13.5 minutes after it was first received. A message that is not JSON, or that its consumer can never process, is parked at once. A parked message stays there until an operator looks at it, purges it or replays it (Configuration). The sync queue has no dead-letter queue of its own: its failures, expired or refused messages included, move to orchestrator.analysis and are handled there.

RabbitMQ Failures¶

If RabbitMQ is unavailable:

The API Gateway answers 503 Service Unavailable; the email is stored
Clients should retry with exponential backoff

Database Failures¶

If PostgreSQL is unavailable, intake fails (nothing can be stored) and status queries for finished jobs fail.

Monitoring Points¶

Each point below names the metrics that answer it.

Queue Metrics¶

The depth of orchestrator.analysis, orchestrator.analysis.sync, orchestrator.dispatch, orchestrator.shadow and orchestrator.benchmark, and the messages in their dead-letter queues (orchestrator.shadow.dead-letter holds the shadow work dropped as the shadow queue overflows): spot_queue_messages{queue, state}, read from RabbitMQ at each scrape (spot_rabbitmq_up says whether the read succeeded)
What became of each message consumed, and how many a second: spot_messages_total{queue, outcome} (acked, retried, parked)
Consumer lag time: no metric yet; a growing ready depth shows it

Processing Metrics¶

Jobs by workflow and outcome, and their verdicts: spot_analyses_total, spot_verdicts_total; jobs being analysed: spot_jobs_running
Job completion time: spot_analysis_duration_seconds
Analyzers answering unknown, error or timeout: spot_analyzer_calls_total{analyzer, status, run}; why, in the logs
Analyzer response time: spot_analyzer_call_duration_seconds
Responders: spot_responder_calls_total; ingests, and how long retrievers wait for their verdicts: spot_ingests_total, spot_ingest_wait_seconds

Database Metrics¶

Query execution time and connection pool usage: no metric yet

Data Retention¶

Emails, their raw copies and analyses:

Kept for 365 days by default, data_retention_days in spot.yaml (Retention)
Purged at the api-gateway's startup and then daily, in batches; the database deletes each email's raw copy, analyses and signals with it

Queue messages:

Quorum queues, persistent messages
A message that keeps failing, does not decode or can never be processed is parked in its queue's dead-letter queue, where it stays until it is purged or replayed

Job records in Redis: expire after 24 hours.

Security Considerations¶

Data in Transit¶

API: HTTPS with TLS 1.2+
RabbitMQ: Internal network only; messages carry ids, never content
PostgreSQL: Internal network only

Data at Rest¶

Raw messages, views, results and signal evidence encrypted with ENCRYPTION_KEY
Database: Encrypted volumes recommended
Queue: Persistent messages on disk

Data Privacy¶

Email content is never logged; reads of a raw message or a part are logged without it
Email content is for analysts and admins
Plugins declare what they send outside SPOT (spot.plugin.egress), and installing a content plugin needs an admin's consent

System Overview - Architecture
Services - Service details
API reference - HTTP endpoints
Operator guide - the dashboard playbook

Data Flow¶

Overview¶

Email Analysis Flow¶

1. Intake¶

Mail Ingest¶

2. Job Processing¶

3. Responders¶

Shadow workflows¶

Benchmark runs¶

4. Status Query¶

5. Cancel¶

Sequence Diagram¶

Analyzer Communication¶

Error Handling¶

Analyzer Failures¶

Failing messages¶

RabbitMQ Failures¶

Database Failures¶

Monitoring Points¶

Queue Metrics¶

Processing Metrics¶

Database Metrics¶

Data Retention¶

Security Considerations¶

Data in Transit¶

Data at Rest¶

Data Privacy¶

Related Documentation¶