49 min read

AML Monitoring on Solana: Logging Transactions with Quicknode Streams

Overview

Anti-Money Laundering (AML) monitoring programs show up most often in crypto businesses that hold customer funds or connect to the banking system. Centralized exchanges, fiat on and off ramps, payment processors, custodial wallets, and stablecoin issuers are commonly regulated as financial institutions, whether as money services businesses registered with FinCEN in the US, virtual asset service providers under the FATF framework, or crypto-asset service providers licensed under MiCA in the EU. Programs in those categories generally involve transaction monitoring, recordkeeping, and suspicious activity reporting.

The DeFi-native cases are less obvious but increasingly common. Tokenized treasuries and other real-world asset platforms onboard investors and then monitor what those investors do onchain. Permissioned lending pools that gate deposits behind KYC run similar monitoring. Teams operating a front-end or a relayer for an otherwise permissionless protocol often screen and monitor as though they were in scope, even where the protocol itself is not.

Monitoring, recordkeeping, and reporting all rest on a transaction record that can be shown to be complete. Each of the three assumes the log holds everything that happened in the window under review, which makes an AML program only as defensible as the record behind it. When an auditor asks what a monitoring system saw at a specific slot, a description of how the data was collected does not answer the question. The record itself has to demonstrate that nothing is missing.

Solana makes that demonstration harder than it first appears. A polling loop can miss a slot, or record one the cluster later abandons, without raising an error in either case. Skipped slots mean a missing slot number does not prove anything went wrong, so the usual completeness check does not apply. The result is a class of gap that is easy to create, difficult to notice, and expensive to explain later.

This guide works through one example of the capture layer that sits underneath an AML monitoring program: a self-correcting log of the transactions touching a given Solana program, screened against a sanctions watchlist as they arrive, built with Streams and Key-Value Store. It is a starting point for the data plumbing, not a finished monitoring system.

Not Legal or Compliance Advice

This guide covers data engineering, not regulatory analysis, and nothing in it is legal advice or a statement that a system built this way is compliant. Which requirements apply to your business, and whether any monitoring system meets them, are determinations for your own compliance team and counsel. Treat what follows as an example to adapt and validate against your own obligations.

What an Incomplete Transaction Record Actually Costs

The transaction record is the deliverable. Auditors do not assess detection logic in the abstract, they sample the log and ask what the system saw at a given moment, and every answer depends on the record being complete and provably so.

A missing block costs more here than it does in ordinary engineering. Normally it is a bug: you notice it, replay it, move on, and the price is a few engineering hours. In a monitoring program, that same missing block leaves a window of activity you cannot show you were watching, and an audit looks at the record rather than at the effort that produced it. So gaps have to be found and closed deliberately, and while one is open the exposure sits with your organization rather than with whichever provider dropped the data.

Polling is not well suited to this. A worker that calls getBlock in a loop, writes rows, and advances a cursor will drop data whenever the process restarts mid-slot or a call times out while the cursor advances anyway, and it will write data that never becomes canonical when it records a slot the cluster later abandons. None of those cases raises an error, so the log looks healthy either way.

On Solana, slots can be skipped, which means a missing slot number is ambiguous by design. It can be entirely normal, or it can be your gap, and the slot number alone does not distinguish them. A simple completeness check of an uninterrupted increasing counter does not work here. Separately, before a block is finalized the cluster can settle on a competing fork, so a transaction you observed and wrote down may not end up on the chain that everyone else agrees on. A record containing a non-canonical transaction is as much of a problem as one missing a canonical transaction, and it is harder to explain.

Solana transactions are irreversible once confirmed, so by the time a flagged transaction reaches your log it is already final. There is no equivalent of a bank declining a wire before it leaves. Detection enables downstream actions like freezing an account, refusing future activity, or filing a report. Only the narrow case of blocking future activity from an already-flagged address is genuinely preventive, and never the transaction that triggered the flag.

What the Capture Layer Has to Get Right

These are the properties this example is built around. They are data requirements rather than compliance requirements, and they drive the product choices in the section that follows.

Ordered Delivery That Labels Corrections

Data must arrive in chain order, and when a delivery is a correction rather than new activity, the pipeline must say so in the payload. A consumer that cannot distinguish new data from corrected data will either discard corrections or double-count them, and both produce a record that is wrong in a way that is difficult to detect afterward.

Automatic Correction When the Chain Changes

Two mechanisms are needed, and they are not substitutes. The first is holding delivery a fixed distance behind the chain tip, which reduces how often you record a slot that later gets displaced. It is probabilistic, so it reduces exposure without eliminating it.

The second is re-delivering the affected range with canonical data when divergence does happen, tagged so the consumer knows to overwrite. Distance from the tip alone leaves you permanently wrong on the cases that slip through.

A Way to Backfill Gaps You Find Later

Automatic correction handles what the chain does to you near the tip. It does nothing for a gap you find out about later: a pipeline that was paused for a day, a monitoring window that predates the system, or a lookback period an auditor asks you to produce. That needs a second, operator-initiated mechanism that replays an explicit slot range through the same screening logic and into the same table, so recovered records are shaped and screened identically to records captured live.

Screening That Does Not Rely on the Program's Own Checks

Assume the program you are monitoring already rejects known-bad addresses onchain. An external check still matters, for two distinct reasons:

• Staleness: An onchain block list is only as current as its last deployment, while a compliance team's own list, sourced from OFAC or a commercial feed, can be updated many times a day. Redeploying a program on every list revision is not practical.
• Circumvention: A motivated actor may have routed around the program's check entirely, through an unflagged wallet, an intermediary hop, or a gap in the program's own enforcement logic.

Both cases point the same way: screen actual transaction data, independent of what the program's logic did. If your only control is the one an adversary is working around, you do not have a control.

Every Screening Result Names the List It Used

A screening result means nothing without the list version that produced it. Sanctions lists change, so "this address was clean when we checked" is only evidence if you can show what the list contained at that moment.

Fixing the Record Without Paging Analysts Twice

These are different problems that look similar. The dataset must be corrected when a slot changes. An analyst must not be paged twice for the same underlying transaction. Solving either one does not solve the other, and conflating them tends to produce a system that either spams analysts or quietly drops corrections.

Independently Audited Infrastructure

Your controls sit on top of a vendor's controls, and an auditor will ask about both. SOC 1, SOC 2, and ISO 27001 attestations are the practical form that question takes.

How Quicknode Products Address These Needs

Two Quicknode products cover most of the properties above, delivering into a database you own:

Streams

Streams delivers blockchain data as an ordered sequence of batches, applies your filter server-side before delivery, and routes the result to a destination you choose: a webhook, S3-compatible storage, PostgreSQL, Azure Blob Storage, or Kafka.

Six of its properties map onto the list above:

• Acknowledged sequential delivery. Streams processes blocks in order and will not advance to the next batch until the current one is confirmed delivered. This is the structural difference from a polling loop, where the cursor advances because your own code said so. Here it advances because the destination acknowledged receipt.
• Failure pauses the Stream. If delivery fails, Streams retries according to your configuration, and after the maximum attempts it pauses the Stream rather than moving on. A broken consumer becomes a stopped pipeline you can see and resume, not a silent hole in the record.
• Latest Block Delay. Delivery can be held a configurable number of slots behind the chain tip. This is the probabilistic half of the correction requirement.
• Restream on reorg. This is the deterministic half. When Streams detects that the chain it delivered no longer matches the canonical chain, it re-delivers the affected range with canonical data and marks the delivery as a correction in the batch metadata.
• Historical range replay. A Stream can be given an explicit start and end slot instead of following the tip, which is how you close a known gap or produce a lookback period. Because the replay runs the same filter into the same destination, recovered records are indistinguishable in shape and screening from live ones.
• Server-side filters. Your screening logic runs inside the pipeline, in JavaScript or Go, so a transaction is evaluated before it reaches any system of yours. This also means the screening result travels with the record instead of being applied afterward, so the result is auditable rather than reconstructed.

Paid Plan Requirements

Every paid plan includes the Restream on reorg option for at-the-tip streams, and it is the mechanism the correction step depends on. Without it, a displaced slot stays in your record uncorrected.

Which destination you can deliver to also depends on your plan. The webhook destination this guide uses is the one available everywhere:

Destination	Available on
Webhook	Every plan, including the free trial
PostgreSQL	Build and above
S3-compatible storage	Build and above
Kafka	Build and above
Azure Blob Storage	Accelerate and above

The number of destinations a single Stream can feed is capped the same way: one on the free trial and Build, up to four on Accelerate, Scale, and Business, and up to six on Enterprise. See the pricing page for the current matrix.

If you do not have an account yet, create a Quicknode account and select a paid plan, then enable the option in your Stream settings.

Key-Value Store

Key-Value Store is managed storage for lists and key-value pairs, reachable two ways: from inside a Streams filter, and from a REST API that works independently of Streams. That dual access is why it fits here rather than a database of your own.

It fills two roles, and both exist because a filter runs with no network access, which makes Key-Value Store the only mutable state it can read:

• The sanctions watchlist, as a list. Because the REST API can update the list without touching the Stream, your compliance team's list can change as often as the feed publishes while the filter code stays untouched. This is the direct answer to the staleness problem that makes an onchain block list insufficient. Lists are scoped to your Quicknode account rather than to one Stream, so the same watchlist serves every Stream and any external service that needs to check membership.
• The list version, as a key-value pair. The filter reads it on every batch and attaches it to each record, so a screening result cites the revision that produced it. The citation is only worth something if you also archive what each revision contained, which the implementation section covers.

Alert deduplication is a third piece of state, but it belongs in the database rather than here. Suppressing a duplicate page has to be decided by the component that knows the write succeeded, and a unique key on the alert does that in the same transaction as the record. Putting it in the filter instead introduces a failure mode where a delivery that gets retried finds the key already set and drops the alert entirely.

A Database You Control

The audit log lands in a database you own. Alerts travel the same path, so there is only ever one route into your systems and everything on it has passed through the reorg handling and the watchlist check.

Owning the destination is also what makes the log queryable for investigations and retainable for as long as your own retention policy calls for, which for compliance records is often measured in years.

Reference Architecture

Two outputs, one pass over the data. The log is passive and comprehensive. The alert is narrow and interruptive. They stay separate because they have different consumers, different retention requirements, and different failure modes.

• Capture: The Stream reads the Solana Block dataset, which delivers full blocks in the shape returned by getBlock. Every block passes through the filter, which keeps only the transactions that invoke the monitored program and discards the rest before anything is delivered. Streams compares each new block's parent hash against the hash of the block it previously delivered, and on a mismatch polls the chain backward to find where the two histories last agreed.

• Screen: For each kept transaction, the filter normalizes the accounts that moved value into per-address records, checks them against the watchlist in one batched lookup, and stamps every record with the list version. One payload leaves the filter carrying four arrays: the blocks in the batch, the matched transactions, the per-address screening results, and any alerts.

• Correct: Every batch tells you which slots it covers, in batch_start_range and batch_end_range. Rather than branching on whether that range moved forward, the ingest service deletes whatever it already holds for the range, along with any slot the delivery names as reorged, and writes the delivery fresh, every time. Streams detects the divergence and re-delivers, but only your code can make the table reflect it.

• Deduplicate: Because a correction re-delivers slots already seen, alert suppression keys on the transaction and address pair rather than on the delivery. Correcting the record and suppressing a duplicate page happen at different layers, on purpose.

There is no indexer to run and no chain-following state machine to maintain, because the ordering and correction guarantees come from the pipeline rather than from code you own.

Implementation

What follows builds the whole thing end to end: a watchlist in Key-Value Store, a Stream configured for ordered and reorg-corrected delivery, a filter that keeps only the transactions touching your program and screens every address that moved value, an ingest service that keeps the tables honest when a slot gets replaced, and a query that proves the record has no holes in it.

The example monitors a single program. Because that address is a constant, it stays hardcoded in the filter, and Key-Value Store is used only for the two things the filter cannot hardcode: the watchlist, which changes on the compliance team's schedule, and the version string that identifies which revision of the list produced a given screening result.

What You Will Need

• A Quicknode account on a paid plan (Restream on reorg and Solana historical replay both require one)
• Your Quicknode Key-Value Store REST API Key
• Node.js (v20.6+, for --env-file support)
• PostgreSQL (v14+) installed locally, as below, or a database on a host like Supabase
• ngrok to expose your local ingest service while you develop
• The address of a Solana program you want to monitor

Set Up the Project

The watchlist sync job and the ingest service share one small Node project. Create it first, so there is a home for the configuration every later step reads:

mkdir solana-aml && cd solana-aml
npm init -y
npm install express@4.21.2 pg@8.13.1

Both services read their configuration from the environment. Three values cover everything they need, so put them in a .env file at the project root:

.env

DATABASE_URL="postgres://localhost:5432/solana_aml"
QN_API_KEY="your-quicknode-api-key"
QN_SECURITY_TOKEN=""

Then keep it out of version control before you put anything real in it:

echo ".env" >> .gitignore

DATABASE_URL names the database you will create in the next section. Leave it as it is if you follow the local install there.

QN_API_KEY is your Quicknode API key, which belongs to your account rather than to any one Stream. Everything that calls Quicknode outbound sends it as an x-api-key header: the watchlist sync job publishing the list to Key-Value Store, and the curl commands that create Streams later on.

QN_SECURITY_TOKEN is the opposite direction, and stays empty for now. Quicknode generates it per Stream, you never send it anywhere, and the ingest service uses it to verify that an incoming webhook really came from your Stream. The Stream that issues it does not exist yet, so you will create the Stream under Write the Stream Filter and fill this in from its Settings tab before running the ingest service.

Node reads that file directly with --env-file, so no dotenv package is needed:

node --env-file=.env sync-watchlist.js

Set Up the Database

Everything in this guide reads and writes one PostgreSQL database, at the connection string you just put in DATABASE_URL. If you already have a database, put its connection string there instead and skip to the next section.

Otherwise, install Postgres and start the server. On macOS with Homebrew:

brew install postgresql@16
brew services start postgresql@16

On Debian or Ubuntu:

sudo apt install postgresql
sudo systemctl start postgresql

Then create the database that DATABASE_URL names:

createdb solana_aml

Confirm you can connect before creating anything in it:

psql solana_aml -c "SELECT version();"

Create the Audit Tables

Five tables, each with one job. Save this as schema.sql:

schema.sql

-- One row per block delivered. This is the completeness spine.
CREATE TABLE blocks (
  slot               BIGINT PRIMARY KEY,
  blockhash          TEXT NOT NULL,
  previous_blockhash TEXT NOT NULL,
  block_time         TIMESTAMPTZ,
  received_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
  corrected_at       TIMESTAMPTZ
);

-- One row per matched transaction, tied to the block it landed in.
CREATE TABLE transactions (
  signature    TEXT    NOT NULL,
  slot         BIGINT  NOT NULL REFERENCES blocks(slot) ON DELETE CASCADE,
  signer       TEXT    NOT NULL,
  fee_lamports BIGINT,
  succeeded    BOOLEAN NOT NULL,
  PRIMARY KEY (signature, slot)
);

-- One row per address that moved value, with the screening result and the
-- list revision that produced it.
CREATE TABLE screened_accounts (
  signature    TEXT    NOT NULL,
  slot         BIGINT  NOT NULL REFERENCES blocks(slot) ON DELETE CASCADE,
  address      TEXT    NOT NULL,
  sol_delta    BIGINT,
  sanctioned   BOOLEAN NOT NULL,
  list_version TEXT    NOT NULL,
  PRIMARY KEY (signature, slot, address)
);

-- Investigations look up an address directly. The primary key above starts
-- with signature, so it cannot serve that query on its own.
CREATE INDEX screened_accounts_address_idx ON screened_accounts (address);

-- Raised alerts. Deliberately NOT tied to blocks: see the ingest service section.
CREATE TABLE alerts (
  alert_key     TEXT PRIMARY KEY,
  signature     TEXT   NOT NULL,
  slot          BIGINT NOT NULL,
  address       TEXT   NOT NULL,
  list_version  TEXT   NOT NULL,
  raised_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
  dispatched_at TIMESTAMPTZ
);

-- Archived watchlist revisions, so a stamped list_version resolves to the
-- exact contents that produced it.
CREATE TABLE watchlist_versions (
  list_version TEXT PRIMARY KEY,
  content_hash TEXT   NOT NULL,
  addresses    TEXT[] NOT NULL,
  published_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

Then apply it:

psql solana_aml -f schema.sql

Five details in there are doing real work:

• blocks gets a row for every block delivered, even blocks with no matching transactions. That is what makes completeness checkable at all. A table of matched transactions alone cannot tell you whether a quiet stretch means "nothing happened" or "we were not watching."
• previous_blockhash is stored, not just the slot. Solana skips slots, so a missing slot number proves nothing. Block hashes chain to each other regardless of how many slots were skipped between them, which is what the completeness check exploits.
• ON DELETE CASCADE on transactions and screened_accounts is the whole correction mechanism. Deleting one blocks row takes its transactions and their screening results with it, so replacing a displaced slot is a single statement.
• alerts has no foreign key and no cascade, on purpose. A page that has already gone out to an analyst is a historical fact, and a reorg does not unsend it. Keeping alerts outside the cascade is what separates correcting the dataset from suppressing a duplicate page, which the two mechanisms above would otherwise conflate.
• watchlist_versions stores the actual addresses, not just the version label. A version string on a record is a pointer, and a pointer to nothing proves nothing. Archiving each revision is what lets you answer "what did that list contain" a year later, and it is what makes the verdicts independently re-derivable in the verification queries.

The screening result lives on screened_accounts rather than on transactions because screening is per address, not per transaction. One transfer produces a row for the sender and a row for the recipient, each carrying its own verdict and list_version.

What these tables do not address is volume over time. They are written to be correct and easy to reason about rather than optimal at scale, and two of them grow steadily for different reasons. blocks grows at whatever rate the chain produces blocks, on the order of 200,000 rows a day no matter how quiet your program is, and it is the cheapest table here. screened_accounts grows with your program's activity multiplied by the number of addresses per transaction, and it is the one that gets large first.

How much that matters depends entirely on how much traffic your monitored program sees, and the honest range runs from trivial to a real engineering problem. Because AML recordkeeping is usually measured in years, it compounds rather than plateaus.

A production deployment will probably want some combination of range partitioning by slot or by month, compression and cheaper storage for older partitions, a retention policy tied to your actual regulatory window, and possibly a more compact representation of screening results than one row per address. Two constraints are worth carrying into that design: narrowing what you screen is a coverage decision rather than a storage optimization, because it makes the absence of a row ambiguous, and watchlist_versions is the one table you cannot prune freely, since dropping a revision invalidates every verdict that cites it.

Sizing, partitioning, and archival are all out of scope here. Work them out against your own volumes and retention obligations before this handles production traffic.

Load the Watchlist into Key-Value Store

The filter runs in a sandbox with no network access, so it cannot call out to your database, an internal service, or a sanctions feed. Key-Value Store is the only mutable state a filter can read, which is what makes it the right home for a list that changes independently of your code.

Two keys live there: the list itself, and a string identifying which revision of the list it currently holds. The filter reads both.

That version string is the part worth being careful about. On its own it is only a label, and a label pointing at contents you no longer have proves nothing when someone asks what the list held at a given slot. So the sync job below does three things together: it derives the version from the contents rather than asserting it, archives the contents under that version, and only then publishes.

Put your feed's addresses in watchlist.json as a flat JSON array of address strings, for example ["Addr1...", "Addr2..."], then run this from the same project as the ingest service:

// sync-watchlist.js
const crypto = require('crypto');
const fs = require('fs');
const { Pool } = require('pg');

const pool = new Pool({ connectionString: process.env.DATABASE_URL });
const KV = 'https://api.quicknode.com/kv/rest/v1';
const WATCHLIST_KEY = 'sanctioned-addresses';
const LIST_VERSION_KEY = 'sanctions-list-version';
const FEED_ID = 'ofac-sdn';

async function kv(method, path, body) {
  const res = await fetch(`${KV}${path}`, {
    method,
    headers: {
      'Content-Type': 'application/json',
      'x-api-key': process.env.QN_API_KEY,
    },
    body: body ? JSON.stringify(body) : undefined,
  });
  const text = await res.text();
  if (!res.ok) throw new Error(`${method} ${path} -> ${res.status} ${text}`);
  return text ? JSON.parse(text) : null;
}

// A missing key is an expected state, not a failure, so 404 comes back as null
// while a 403 still throws. Collapsing the two is what would turn a bad API key
// into a watchlist that looks published and is not.
async function kvGet(path) {
  const res = await fetch(`${KV}${path}`, {
    headers: { 'x-api-key': process.env.QN_API_KEY },
  });
  if (res.status === 404) return null;
  const text = await res.text();
  if (!res.ok) throw new Error(`GET ${path} -> ${res.status} ${text}`);
  return text ? JSON.parse(text) : null;
}

async function main() {
  // Replace with your feed parser. It only has to return an array of addresses.
  const raw = JSON.parse(fs.readFileSync('watchlist.json', 'utf8'));
  const addresses = [...new Set(raw)].sort();

  // The version is DERIVED from the contents, so it cannot disagree with them.
  // A hand-written date can: someone bumps it without changing the list, or
  // changes the list and forgets to bump it. A content hash cannot lie.
  const contentHash = crypto
    .createHash('sha256')
    .update(addresses.join('\n'))
    .digest('hex');
  const today = new Date().toISOString().slice(0, 10);
  const listVersion = `${FEED_ID}-${today}.${contentHash.slice(0, 8)}`;

  // What Key-Value Store actually holds right now. Reconciling against this
  // rather than against the archive is what lets a half-finished run heal.
  const liveItems = (await kvGet(`/lists/${WATCHLIST_KEY}`))?.data?.items ?? null;
  const liveVersion = (await kvGet(`/values/${LIST_VERSION_KEY}`))?.data?.value ?? null;

  const client = await pool.connect();
  try {
    const { rows: [previous] } = await client.query(
      `SELECT list_version, content_hash, addresses
       FROM watchlist_versions ORDER BY published_at DESC LIMIT 1`,
    );

    // Skip only when the archive, the live list, and the live version all
    // agree. Checking the archive alone would read a run that failed partway
    // as a completed one and never retry it.
    const listMatches = liveItems
      && liveItems.length === addresses.length
      && [...liveItems].sort().join('\n') === addresses.join('\n');

    if (previous?.content_hash === contentHash
        && liveVersion === previous.list_version
        && listMatches) {
      console.log(`no change, still ${previous.list_version}`);
      return;
    }

    // 1. Archive first, so the version resolves to real contents before any
    //    record can cite it.
    await client.query(
      `INSERT INTO watchlist_versions (list_version, content_hash, addresses)
       VALUES ($1, $2, $3)
       ON CONFLICT (list_version) DO NOTHING`,
      [listVersion, contentHash, addresses],
    );

    // 2. Bring the live list in line, as a diff against what the list actually
    //    contains. PATCH would 404 on a list that was never created.
    let addItems = addresses;
    let removeItems = [];

    if (liveItems) {
      const before = new Set(liveItems);
      const after = new Set(addresses);
      addItems = addresses.filter((a) => !before.has(a));
      removeItems = liveItems.filter((a) => !after.has(a));
      if (addItems.length || removeItems.length) {
        await kv('PATCH', `/lists/${WATCHLIST_KEY}`, { addItems, removeItems });
      }
    } else {
      await kv('POST', '/lists', { key: WATCHLIST_KEY, items: addresses });
    }

    // 3. Publish the version last. See the note on ordering below.
    await kv('POST', '/values', { key: LIST_VERSION_KEY, value: listVersion });

    console.log(
      `published ${listVersion}:`,
      `+${addItems.length} -${removeItems.length}, ${addresses.length} total`,
    );
  } finally {
    client.release();
    await pool.end();
  }
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

Run it whenever your feed publishes:

node --env-file=.env sync-watchlist.js

Four things in there are deliberate.

The job reconciles against Key-Value Store, not against the archive. The archive commits on its own, so a run that fails at the KV call leaves a version row behind with nothing published. Deciding whether there is work to do from the archive alone would read that as a finished sync and skip it on every later run, leaving the list permanently empty while the job reports no change. Reading the live list and version first means the next run repairs it.

The version is a content hash, not a date. ofac-sdn-2026-07-30.a3f9c1b2 is derived from the sorted addresses, so it cannot drift from what it describes, and re-running the job when nothing changed is a no-op instead of a spurious new version. The date prefix is there for humans; the suffix is the part that is verifiable.

The archive is written before the list is published. Do it the other way and there is a window where records cite a version that resolves to nothing.

The version value is published last. Key-Value Store has no transaction spanning a list and a value, so there is a brief window where the two disagree. Publishing the version last means that during the window the screening is current while the label lags, which is the safer failure: you would rather catch a newly listed address and have a bookkeeping discrepancy than screen against a stale list. Either way the discrepancy is detectable, and the coherence query under Verify the Record Is Complete is what finds it.

Note the casing difference between the two interfaces, which is easy to trip over: the REST API takes addItems and removeItems, while the qnUpsertList method inside a filter takes add_items and remove_items.

Because lists are scoped to your Quicknode account rather than to one Stream, this same watchlist serves every Stream you run and any external service that needs to check membership through the REST API.

Seed the list from a real source. Populate watchlist.json from an actual feed, whether that is a parser for the OFAC SDN list or a commercial provider, and never from addresses you have not sourced from one. To smoke-test the alert path before you have a feed wired up, put in an address you expect to see in your own matched transactions and remove it afterward. Treat anything you add as a claim about a real party, because that is what a sanctions hit in your log will read as later.

Write the Stream Filter

Create your Stream in the dashboard, pick Solana and Mainnet, select the Block dataset, and choose "Modify the payload before streaming". Replace the default main function with this:

// Solana Block dataset, jsonParsed encoding. Requires a batch size of 1
// (see the notes below).
const MONITORED_PROGRAM = 'jupoNjAxXgZ4rjzxzPMP4oxduvQsQtZzyknqvzYNrNu';
const WATCHLIST_KEY = 'sanctioned-addresses';
const LIST_VERSION_KEY = 'sanctions-list-version';

async function main(stream) {
  const meta = stream.metadata || {};
  const startSlot = meta.batch_start_range;
  const endSlot = meta.batch_end_range;

  // One read per batch. Every record carries this, so a screening result
  // cites the exact revision it was made against. The revision itself is
  // archived in watchlist_versions, which is what the citation resolves to.
  // If the key is missing we stamp 'unversioned' rather than throwing, so a
  // misconfiguration degrades into rows the coherence query flags instead of
  // a paused Stream.
  const listVersion = (await qnLib.qnGetValue(LIST_VERSION_KEY)) || 'unversioned';

  const blocks = [];
  const matched = [];

  for (const block of stream.data) {
    blocks.push({
      slot: startSlot,
      blockhash: block.blockhash,
      previousBlockhash: block.previousBlockhash,
      blockTime: block.blockTime,
    });

    for (const tx of block.transactions || []) {
      const keys = (tx.transaction?.message?.accountKeys || []).map((k) => k.pubkey || k);
      if (!keys.includes(MONITORED_PROGRAM)) continue;

      matched.push({
        signature: tx.transaction.signatures[0],
        signer: keys[0],
        feeLamports: tx.meta?.fee ?? null,
        succeeded: tx.meta?.err === null,
        accounts: accountsThatMovedValue(tx, keys),
      });
    }
  }

  // Every address in the batch screened in a single round trip. The result is
  // index-aligned with the addresses we sent.
  const addresses = [...new Set(matched.flatMap((m) => m.accounts.map((a) => a.address)))];
  const hits = addresses.length
    ? await qnLib.qnContainsListItems(WATCHLIST_KEY, addresses)
    : [];
  const sanctioned = new Set(addresses.filter((_, i) => hits[i]));

  const transactions = [];
  const screened = [];
  const alerts = [];

  for (const m of matched) {
    transactions.push({
      signature: m.signature,
      slot: startSlot,
      signer: m.signer,
      feeLamports: m.feeLamports,
      succeeded: m.succeeded,
    });

    for (const account of m.accounts) {
      const isSanctioned = sanctioned.has(account.address);

      screened.push({
        signature: m.signature,
        slot: startSlot,
        address: account.address,
        solDelta: account.solDelta,
        sanctioned: isSanctioned,
        listVersion,
      });

      if (isSanctioned) {
        alerts.push({
          alertKey: `${m.signature}:${account.address}`,
          signature: m.signature,
          slot: startSlot,
          address: account.address,
          listVersion,
        });
      }
    }
  }

  // Always return a payload, never null. On a block with no matching
  // transactions, the arrays are empty but `blocks` still carries the hash
  // link that the completeness check depends on.
  return {
    batch: {
      startSlot,
      endSlot,
      reorgedSlots: meta.blocks_reorged || [],
    },
    blocks,
    transactions,
    screened,
    alerts,
  };
}

// Addresses whose balance actually changed. SOL movements come from the
// balance arrays. Token movements are reported against a token account, so
// the owner is the party to screen, not the token account itself.
function accountsThatMovedValue(tx, keys) {
  const pre = tx.meta?.preBalances || [];
  const post = tx.meta?.postBalances || [];
  const moved = new Map();

  keys.forEach((address, i) => {
    const delta = (post[i] ?? 0) - (pre[i] ?? 0);
    if (delta !== 0) moved.set(address, delta);
  });

  const tokenAccounts = new Map();
  for (const balance of tx.meta?.preTokenBalances || []) {
    tokenAccounts.set(balance.accountIndex, {
      owner: balance.owner,
      before: balance.uiTokenAmount?.amount ?? '0',
      after: '0',
    });
  }
  for (const balance of tx.meta?.postTokenBalances || []) {
    const entry = tokenAccounts.get(balance.accountIndex)
      || { owner: balance.owner, before: '0', after: '0' };
    entry.owner = entry.owner || balance.owner;
    entry.after = balance.uiTokenAmount?.amount ?? '0';
    tokenAccounts.set(balance.accountIndex, entry);
  }
  for (const entry of tokenAccounts.values()) {
    if (entry.owner && entry.before !== entry.after && !moved.has(entry.owner)) {
      moved.set(entry.owner, null);
    }
  }

  return [...moved].map(([address, solDelta]) => ({ address, solDelta }));
}

Save the same code locally as filter.js in your project directory. The dashboard holds the copy the Stream runs, but the REST API calls under Configure the Stream and Close a Known Gap both read it from disk.

Set the "Test Block" field to a recent slot and click "▶️ Run Test". With the SPL Token program above you will match most blocks, which makes for a useful smoke test. Swap in your own program address once you have seen it work.

A test populates stream.metadata, so batch_start_range resolves to the slot you tested against and startSlot and endSlot come back equal to it. The one part a test cannot exercise is the correction path: blocks_reorged is only present on a delivery that actually replaces chain history, so reorgedSlots is an empty array in every test.

This Filter Assumes jsonParsed Encoding

To keep the code short, this filter reads accounts straight out of message.accountKeys and trusts that one array holds every account in the transaction, with indices that line up with preBalances and postBalances. That is true under jsonParsed, where address lookup table accounts are merged into accountKeys and each entry is an object carrying a pubkey.

The choice has consequences worth understanding before this handles production traffic. Under json encoding, accountKeys holds only the statically declared keys, as bare strings, and lookup table accounts arrive separately in meta.loadedAddresses, ordered after the static keys as writable then readonly. Code written against accountKeys alone would then miss every program reached through a lookup table and never screen the addresses loaded from one. Neither failure raises an error, and on a recent Mainnet block that gap covered more than half the transactions touching the SPL Token program, which in a screening system is the worst way for something to be wrong.

If you stream any encoding other than jsonParsed, reassembling the full account list and keeping it aligned with the balance arrays is yours to handle.

Seven decisions in that filter are worth spelling out, because each one is a place the obvious version of this code goes wrong.

It never returns null, even when no transaction matched. Most filters return null when nothing matches, which skips delivery entirely. Here that breaks the completeness check, so it is worth being precise about what the filter returns on a quiet block: an empty transactions array, but a populated blocks array. The block record is the hash link, and that is the thing you cannot afford to drop.

To see why, take three consecutive blocks where only the first and third touch your program. Skip the middle one and your table holds slot 100 (hash A) and slot 102 (parent hash B). Those rows are no longer adjacent onchain, so the continuity query compares B against A, finds they differ, and reports a gap at slot 102. Nothing is actually missing. Because most blocks will not touch your program, that false alarm fires on nearly every quiet stretch, and a completeness alarm that goes off constantly is one you stop reading.

Storing every block does not pollute the log, because blocks and transactions are separate tables: the completeness spine stays dense while the transaction log stays sparse. The cost is delivery volume rather than API credits, since credits are consumed per block processed regardless of what the filter returns. On Solana that is roughly 2.5 small deliveries per second. If that traffic is not worth it for your use case, the tradeoff you are accepting is giving up independent completeness verification and relying on Streams' sequential delivery instead.

The slot comes from metadata.batch_start_range, not from the block body. Solana's getBlock response does not contain its own slot number. The tempting substitute is parentSlot + 1, which is only correct when no slot was skipped immediately before this block, and silently off by one or more when a slot was. The batch range is authoritative. This is also why the batch size has to be 1: at that size batch_start_range and batch_end_range are the same slot, and each delivery maps to exactly one block.

Matching on accountKeys catches CPI. Checking instructions[].programId only finds top-level invocations, so a transaction that reaches your program through an intermediary would slip past. Every program invoked anywhere in a transaction, including through a cross-program invocation, has to appear in that transaction's account keys, so this check covers both cases.

The batch range and reorg flags are copied into the payload body. Streams also sends metadata as HTTP headers, but once a filter shapes the payload the body is whatever your filter returned. Putting the range and blocks_reorged inside the payload is what gives you the in-band completeness signal: the delivery says what it covers and whether it is a correction, without your ingest service depending on header parsing.

Screening happens in one batched lookup per delivery, not one per address. The filter collects every address in the batch, deduplicates them, and makes a single qnContainsListItems call that returns a boolean array aligned to what it sent. The docs are emphatic about this, and the reason is latency: a per-address call inside a nested loop turns one round trip into hundreds, and the Stream cannot advance until the filter returns.

Screening runs on balance changes, not on instruction parsing. The filter screens any address whose SOL balance moved, plus the owner of any token account whose balance moved. Working from the balance arrays in meta rather than decoding instructions means it does not need to understand your program's instruction layout, and it cannot be evaded by an unusual call path: if value moved, the balance changed, and the address is screened. Token balances are reported against the token account rather than the wallet, which is why the code resolves owner instead of screening the token account address, since the token account is not the party a sanctions list names.

The screening verdict is computed before delivery and travels with the record. This is the difference between an auditable result and a reconstructed one. Nothing downstream has to re-derive whether an address was listed, and nothing has to guess which revision of the list applied, because both are already on the row.

Build the Ingest Service

Create server.js in the project you set up earlier:

// server.js
const crypto = require('crypto');
const express = require('express');
const { Pool } = require('pg');

const pool = new Pool({ connectionString: process.env.DATABASE_URL });
const SECURITY_TOKEN = process.env.QN_SECURITY_TOKEN;

const app = express();
app.use(express.raw({ type: 'application/json', limit: '50mb' }));

function verifySignature(req) {
  const nonce = req.get('X-QN-Nonce');
  const timestamp = req.get('X-QN-Timestamp');
  const signature = req.get('X-QN-Signature');
  if (!nonce || !timestamp || !signature) return false;

  const expected = crypto
    .createHmac('sha256', SECURITY_TOKEN)
    .update(nonce + timestamp + req.body.toString('utf8'))
    .digest('hex');

  const a = Buffer.from(expected, 'hex');
  const b = Buffer.from(signature, 'hex');
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

app.post('/streams', async (req, res) => {
  if (!verifySignature(req)) {
    return res.status(401).send('invalid signature');
  }

  const { batch, blocks, transactions, screened, alerts } = JSON.parse(
    req.body.toString('utf8'),
  );
  const client = await pool.connect();

  try {
    await client.query('BEGIN');

    // reorgedSlots is the authoritative signal that this delivery replaced
    // chain history. A non-empty delete on its own only means we had seen
    // these slots before, which is also true of an ordinary retry.
    const reorgedSlots = batch.reorgedSlots || [];
    const isCorrection = reorgedSlots.length > 0;

    // Anything already stored in this slot range is superseded by what just
    // arrived. Removing it first is what makes a correction a correction:
    // rows that existed only on the displaced block go away, instead of
    // sitting in the table next to the canonical ones. The reorged slots are
    // deleted as well as the delivered range, because the new canonical chain
    // may have no block at all for a slot that previously had one, and a slot
    // that never gets re-delivered would otherwise keep its displaced row.
    const { rowCount: rewritten } = await client.query(
      'DELETE FROM blocks WHERE slot BETWEEN $1 AND $2 OR slot = ANY($3::bigint[])',
      [batch.startSlot, batch.endSlot, reorgedSlots],
    );

    for (const b of blocks) {
      await client.query(
        `INSERT INTO blocks (slot, blockhash, previous_blockhash, block_time, corrected_at)
         VALUES ($1, $2, $3, to_timestamp($4), CASE WHEN $5 THEN now() END)
         ON CONFLICT (slot) DO NOTHING`,
        [b.slot, b.blockhash, b.previousBlockhash, b.blockTime, isCorrection],
      );
    }

    for (const t of transactions) {
      await client.query(
        `INSERT INTO transactions (signature, slot, signer, fee_lamports, succeeded)
         VALUES ($1, $2, $3, $4, $5)
         ON CONFLICT (signature, slot) DO NOTHING`,
        [t.signature, t.slot, t.signer, t.feeLamports, t.succeeded],
      );
    }

    for (const s of screened) {
      await client.query(
        `INSERT INTO screened_accounts
           (signature, slot, address, sol_delta, sanctioned, list_version)
         VALUES ($1, $2, $3, $4, $5, $6)
         ON CONFLICT (signature, slot, address) DO NOTHING`,
        [s.signature, s.slot, s.address, s.solDelta, s.sanctioned, s.listVersion],
      );
    }

    // The alerts insert IS the deduplication gate. alert_key is the primary
    // key, so a signature and address pair that has already alerted conflicts
    // and returns nothing. Only rows that did not exist come back, and only
    // those become a page.
    const toDispatch = [];
    for (const a of alerts) {
      const { rows } = await client.query(
        `INSERT INTO alerts (alert_key, signature, slot, address, list_version)
         VALUES ($1, $2, $3, $4, $5)
         ON CONFLICT (alert_key) DO NOTHING
         RETURNING alert_key`,
        [a.alertKey, a.signature, a.slot, a.address, a.listVersion],
      );
      if (rows.length > 0) toDispatch.push(a);
    }

    await client.query('COMMIT');

    if (isCorrection) {
      console.log(
        `corrected slots ${batch.startSlot}-${batch.endSlot}`,
        `reorged: ${JSON.stringify(reorgedSlots)}`,
      );
    } else if (rewritten > 0) {
      console.log(`re-delivery of slots ${batch.startSlot}-${batch.endSlot}, rewrote in place`);
    }

    // Dispatch only after the commit. Paging an analyst about a record you
    // failed to store is worse than paging them a moment later.
    for (const a of toDispatch) {
      try {
        await dispatchAlert(a);
        await pool.query(
          'UPDATE alerts SET dispatched_at = now() WHERE alert_key = $1',
          [a.alertKey],
        );
      } catch (err) {
        // The row is already committed, so the alert is not lost. Sweep for
        // rows where dispatched_at IS NULL and retry them out of band.
        console.error('alert dispatch failed, row retained', a.alertKey, err);
      }
    }

    res.sendStatus(200);
  } catch (err) {
    // If the connection itself is what failed, ROLLBACK throws too. Swallow
    // that so the 500 below still goes out instead of the request hanging
    // until the delivery times out.
    await client.query('ROLLBACK').catch(() => {});
    console.error('ingest failed, refusing to acknowledge', err);
    // Anything other than 2xx makes Streams retry and then pause. Never
    // acknowledge a delivery you did not actually write.
    res.sendStatus(500);
  } finally {
    client.release();
  }
});

// Replace with your paging path: PagerDuty, Slack, a case management queue.
// Keep it out of the database transaction.
async function dispatchAlert(alert) {
  console.log(
    `🚨 SANCTIONS HIT ${alert.address} in ${alert.signature}`,
    `at slot ${alert.slot} (list ${alert.listVersion})`,
  );
}

app.listen(3000, () => console.log('ingest listening on http://localhost:3000'));

Fill QN_SECURITY_TOKEN in your .env with the security token from the Stream's Settings tab, then run it:

node --env-file=.env server.js

Then expose it:

ngrok http 3000

The interesting part of that service is how little of it there is. One DELETE followed by inserts, inside one transaction, handles three situations that would otherwise each need their own code path:

• A reorg correction. Streams re-delivers the affected slots with canonical data. The delete drops the displaced version, including any transaction that existed only on the old block, and the inserts write the new one. An upsert alone cannot do this, because an upsert has no way to remove a row that the correction no longer contains. Deleting the slots named in reorgedSlots as well as the delivered range covers the case where the new canonical chain has no block for a slot that previously had one, which is never re-delivered and so would otherwise keep its displaced row forever.
• A retried delivery. If your service writes successfully but the acknowledgement never reaches Streams, the same batch arrives again. Deleting the range and rewriting it produces an identical table, so the duplicate is harmless. That is the dedup requirement, satisfied by making writes idempotent rather than by tracking what has been seen.
• An overlapping replay. A historical Stream covering slots you already have rewrites them in place instead of colliding with them, which is what makes the gap-closing replay safe to run without checking first.

Note that the delete and the corrected_at stamp are driven by different signals, and conflating them is an easy mistake. The delete runs on every delivery, because that is what makes the write idempotent. The corrected_at timestamp is set only when reorgedSlots is non-empty, because that is the only case where chain history actually changed. Keying the timestamp off the delete instead would stamp "corrected" on every ordinary retry and leave you unable to tell a network hiccup from a reorg when reviewing the log later.

Alerts sit outside all of that, which is the point of keeping them in their own table with no cascade. The blocks delete rewrites the dataset freely, over and over if it has to, while alerts accumulates. Because alert_key is the primary key, the insert itself does the suppression: a signature and address pair that has already paged conflicts, RETURNING yields nothing, and no second page goes out. Correcting the record and suppressing a duplicate page therefore happen at different layers and cannot interfere with each other.

Two consequences worth stating plainly. A reorg that removes a transaction does not unsend an alert that already went out, and it should not: the alert is a record of what you acted on, and the alerts row keeps its slot so you can reconcile it against blocks and screened_accounts afterward. This is one of the reasons to run a meaningful Latest Block Delay, since the cheapest way to avoid paging on a transaction that gets displaced is not to see it that early. And because dispatch happens after the commit rather than inside it, a paging outage leaves you with a committed alert whose dispatched_at is still NULL, which is a query you can sweep, rather than a lost page.

Deciding Where the Alert Goes

dispatchAlert is deliberately left as a stub, because this is the point where a data pipeline stops and a compliance process starts. The pipeline's job ends at producing a durable, deduplicated hit. What happens to that hit is your decision, and it depends on your alert volume, how quickly one has to be dispositioned, and whether your procedures require a record of who reviewed it.

Some options, roughly from most to least interruptive:

• On-call paging, through something like PagerDuty or Opsgenie. Appropriate only when someone genuinely has to act within minutes. Remember that the transaction is already final, so whatever urgency exists comes from the downstream action, such as freezing an account or cutting off future activity, rather than from the transaction itself.
• A monitored chat channel in Slack or Teams. Low friction and common, but it leaves no record of who looked at what, which tends to be exactly what an examiner asks about.
• Case management, either a general ticketing system or a purpose-built AML case tool. More work to set up and usually the closest fit, because it produces an assignment, an investigator, and a documented disposition for each hit. That artifact is generally what a regulator wants to see, more than the notification that preceded it.
• A queue or event bus such as SQS, Kafka, or Pub/Sub. Worth it when several systems need the same hit, or when you want the option to replay alerts into a new consumer later.
• Nothing pushed at all. At low volumes this is defensible. The alerts table is already a work queue, and a scheduled review of rows where dispatched_at IS NULL may satisfy the procedure on its own. If that is the choice, write it down as the procedure rather than leaving it as an omission.

Whatever you choose, two properties of the shape above are worth keeping. Leave the routing outside the database transaction, so a paging outage can never roll back a stored record. And treat the alerts table as the source of truth rather than the notification, so that changing destinations is a one-function change and no hit is ever contingent on a delivery that may not have happened.

One step usually belongs here that this example does not attempt: joining the flagged address to your own KYC records before routing, so the alert arrives naming an internal customer rather than a bare address. That mapping lives in your systems rather than onchain, which is why it sits outside this pipeline.

Returning 500 is the correct behavior. It is tempting to catch the error, log it, and return 200 so the pipeline keeps moving. Do not. A 2xx tells Streams the slot is safely stored and lets it advance, which converts a database problem into a permanent hole in the record. Returning 500 makes Streams retry and then pause the Stream, leaving you with a stopped pipeline you can see and resume.

Configure the Stream

Back in your Stream configuration, set the destination to Webhook with your ngrok URL plus the path, for example https://your-subdomain.ngrok-free.app/streams. Then set these options:

Setting	Value	Why
Dataset	Block	Full blocks, as returned by getBlock
Batch size	1	Makes batch_start_range the block's slot
Elastic batch	Off	Keeps the batch size at 1 near the tip
Restream on reorg	On	The deterministic half of the correction requirement
Latest Block Delay	32	The probabilistic half, roughly 13 seconds behind the tip
Compression	None	One less thing to handle in the ingest service

Click "▶️ Test Destination", then create the Stream. Rows should start appearing in blocks within a few seconds.

Verify the Record Is Complete

This is the query that turns the log into evidence. Every block's previous_blockhash should equal the blockhash of the block before it:

SELECT slot AS break_at_slot,
       previous_blockhash AS expected_parent,
       prior_hash         AS actual_prior_block
FROM (
  SELECT slot,
         previous_blockhash,
         LAG(blockhash) OVER (ORDER BY slot) AS prior_hash
  FROM blocks
) linked
WHERE prior_hash IS NOT NULL
  AND previous_blockhash <> prior_hash
ORDER BY slot;

No rows means the blocks you hold form an unbroken chain, with nothing missing between the first and last. Any row is the far edge of a gap: the block at break_at_slot expected a parent you do not have.

This works precisely where a slot-number check does not. Skipped slots produce no block at all, so consecutive blocks stay hash-linked across them and a jump in slot numbers is not evidence of anything. Hash continuity ignores slot numbering entirely and answers the question you actually care about, which is whether you are holding every block that exists.

Two things this query does not tell you, worth being clear about. It cannot detect a gap at either end of the range, since there is nothing on the far side to link to, so pair it with the slot your Stream started at and the current chain tip. And it says nothing about whether individual transactions inside a block were filtered correctly, which is a separate concern from completeness.

Run it on a schedule and alert on any row. It is cheap, and it is the closest thing to a direct answer when someone asks what your monitoring system saw at a given slot.

Three more queries are worth having on hand. The first answers the screening-provenance question directly, which is what an auditor sampling your log will actually ask. Because the cited revision is archived, the query does not just report the stored verdict, it re-derives it:

-- What did we conclude about this address, against which revision, and does
-- that revision actually support the conclusion?
SELECT s.slot, s.signature, s.address, s.sanctioned, s.list_version,
       v.content_hash,
       (s.address = ANY (v.addresses)) AS listed_in_that_revision,
       b.block_time, b.corrected_at
FROM screened_accounts s
JOIN blocks b USING (slot)
LEFT JOIN watchlist_versions v ON v.list_version = s.list_version
WHERE s.address = '9kwU8PYhsmRfgS3nwnzT3TvnDeuvdbMAXqWsri2X8rAU'
ORDER BY s.slot;

The second is the coherence check, and it is the one to run on a schedule. It catches both ways the screening state can go wrong: a verdict that the archived list does not support, and a record citing a revision that was never archived at all:

SELECT s.slot, s.signature, s.address, s.sanctioned, s.list_version,
       CASE WHEN v.list_version IS NULL THEN 'no archived revision'
            ELSE 'verdict disagrees with archived list' END AS problem
FROM screened_accounts s
LEFT JOIN watchlist_versions v ON v.list_version = s.list_version
WHERE v.list_version IS NULL
   OR s.sanctioned <> (s.address = ANY (v.addresses))
ORDER BY s.slot;

Zero rows means every verdict in the table can be independently reproduced from an archived list. That is a stronger statement than "we stamped a version on it," and it is the one that survives someone deliberately trying to poke holes in your process. Non-zero rows point at a specific, fixable problem: usually either the publish-ordering window in the watchlist sync, or rows stamped unversioned because the filter ran before sanctions-list-version existed. Both are resolved by replaying the affected slots.

The third finds alerts that were recorded but never delivered, which is the recovery path for a paging outage:

SELECT alert_key, address, signature, slot, list_version, raised_at
FROM alerts
WHERE dispatched_at IS NULL
ORDER BY raised_at;

Storing list_version per row rather than resolving it globally is what makes this approach work. When the list changes, old records keep the verdict they were given at the time, so a newly listed address does not retroactively rewrite what you claim to have known. Re-screening a past window against a newer list is then a deliberate gap-closing replay rather than a silent side effect.

Close a Known Gap

When the continuity check finds a break, or when you need a lookback window that predates the Stream, create a second Stream over the explicit slot range and point it at the same webhook:

curl -X POST "https://api.quicknode.com/streams/rest/v1/streams" \
  -H "Content-Type: application/json" \
  -H "x-api-key: $QN_API_KEY" \
  -d "{
    \"name\": \"solana-aml-replay-436190000-436191000\",
    \"network\": \"solana-mainnet\",
    \"dataset\": \"block\",
    \"region\": \"usa_east\",
    \"filter_function\": \"$FILTER_B64\",
    \"start_range\": 436190000,
    \"end_range\": 436191000,
    \"dataset_batch_size\": 1,
    \"elastic_batch_enabled\": false,
    \"destination\": \"webhook\",
    \"destination_attributes\": {
      \"url\": \"https://your-subdomain.ngrok-free.app/streams\",
      \"compression\": \"none\",
      \"max_retry\": 3,
      \"retry_interval_sec\": 5,
      \"post_timeout_sec\": 30
    },
    \"status\": \"active\"
  }"

Same filter, same destination, same tables, so recovered records are identical in shape to records captured live and there is no separate backfill path to reason about. Because the ingest service rewrites any range it is handed, you can run a replay over slots you partly hold without checking which ones first. fix_block_reorgs is omitted here because a fixed historical range is already final. The Stream ends on its own at end_range.

One thing to be deliberate about: a replay screens against the watchlist as it stands now, not as it stood when those slots were originally produced. The list_version written on the recovered rows will be today's, which is correct and is exactly what you want when the reason for the replay is a newly listed address. It does mean a replay over slots you already hold is a re-screening, not just a repair, so the previous verdicts for that range are overwritten. If you need the earlier verdicts preserved, snapshot screened_accounts for the range before replaying it.

Re-run the continuity query afterward. The break should be gone.

Solana Replay Requires a Paid Plan

Free Trial accounts can only create Solana Streams that follow the tip of the chain. Historical slot ranges, which is what gap closure depends on, are available on paid plans. Streams bills API credits per block processed, drawn from the same pool as RPC, and filtering does not reduce that cost because the block still has to be fetched and processed. Estimate a replay range with the calculator in the Streams billing documentation before starting one.

Wrapping Up

What this architecture produces is a Solana transaction log that self-corrects when the chain changes underneath it, and a sanctions hit that fires on actual transaction data rather than trusting a program's own enforcement to have worked. The properties that make it dependable are unglamorous: re-delivery is idempotent, every screening result carries the list version that produced it, corrections leave a timestamp behind, and a failed write stops the pipeline instead of skipping a slot.

That is one layer of a monitoring program and not the whole of one. Alert thresholds, case management, investigator workflows, reporting, retention schedules, and the policies behind all of them sit outside this pipeline, and confirming that the finished system meets your obligations is work only you and your compliance team can do.

An auditor will also ask about the infrastructure underneath your controls. Quicknode maintains SOC 1 Type 2, SOC 2 Type 2, and ISO/IEC 27001 certifications. Details are on the security page.

Next Steps

• Decide where the watchlist comes from, whether that is a parser for the OFAC SDN list or a commercial provider, and how often it syncs
• Plan how flagged addresses join against your KYC records, so an alert arrives with an internal identity attached
• Read How to Stream Solana Program Data for other ways to shape a Solana Stream around a program, including widening capture to several programs at once

Have questions or want to talk through a compliance data pipeline? Join our Discord, follow @Quicknode, or contact us directly.

Frequently Asked Questions

Does Solana have chain reorgs, and why does AML monitoring need to handle them?

Solana does not reorganize finalized blocks, but before finality the cluster can settle on a competing fork, so a block observed near the tip may not end up on the canonical chain. Slots can also be skipped, which makes a missing slot number ambiguous: it can be normal or it can be a gap in your record. For AML monitoring both cases matter, because a record containing a non-canonical transaction is as much of a problem as one missing a canonical transaction. Streams detects the divergence by comparing a new block's parent hash against the hash it previously streamed, then re-delivers the affected range tagged with reorgs and blocks_reorged metadata.

Why screen transactions externally if the program already blocks sanctioned addresses onchain?

Two reasons. An onchain block list is only as current as its last deployment, while an external list in Key-Value Store can be updated as often as your sanctions feed publishes. And an adversary may have routed around the program's check entirely, through an unflagged wallet, an intermediary hop, or a gap in the program's own logic. Screening actual transaction data is independent of whether the program's enforcement worked. A control that shares a failure mode with the thing it is checking is not a control.

Will a restreamed slot alert my compliance team twice about the same transaction?

No, as long as alert deduplication is kept separate from data correction. This approach gives each alert a key of signature and address, makes that the primary key of an alerts table that is deliberately excluded from the reorg cascade, and pages only when the insert actually creates a row. Correction rewrites the dataset as often as it needs to while alerts accumulate, so the two cannot interfere. Deduplication does not undo an alert on a transaction that a reorg later removed, which is one reason to run a meaningful keep_distance_from_tip.

What does AML transaction monitoring on Solana cost to run?

Streams bills API credits per block processed, using a multiplier that depends on the network and dataset, and it draws from the same credit pool as RPC. Filters do not reduce that cost, because the block still has to be fetched and processed to apply the filter logic, though they substantially reduce bandwidth and downstream storage. Because Solana produces blocks quickly, estimate before activating using the calculator in the Streams billing documentation. The Restream on reorg option this pipeline depends on requires a paid plan.

Does logging Solana transactions this way make my system AML compliant?

No. This covers one piece of the data layer: capturing a complete, self-correcting transaction record and screening it against a watchlist. Whether a monitoring system satisfies an AML obligation depends on your own requirements, policies, thresholds, investigation and reporting workflows, and retention rules, none of which a data pipeline decides. Treat this as an example to adapt, and confirm the finished system with your compliance team and counsel.

Can I use a database destination directly instead of a webhook and ingest service?

Yes. Streams delivers to webhooks, S3-compatible storage, PostgreSQL, Azure Blob Storage, and Kafka, so a database can receive records with no service of your own in between. The tradeoff is correction handling: a direct destination leaves you relying on an upsert to absorb re-delivered rows, while a webhook and a small ingest service let you remove superseded rows, record when the correction was applied, and dispatch alerts inside one transaction.