When a new video shows up in one of our regional crawlers, three things need to happen almost immediately: the SQLite FTS5 search index for that region needs a new row, the discovery ranking cache needs to be invalidated, and the sitemap generator needs to know a URL was born. For a long time we did all of this inline, inside the same cron process that fetched the video. It worked until it didn't. A slow FTS5 rebuild would stall the fetch loop, a sitemap write would fail silently, and a crash halfway through meant one region had the video indexed and another didn't. The fetch and the fan-out were fused together, and every failure was a partial failure.
The fix was to stop treating "a video was discovered" as a function call and start treating it as an event. At TrendVidStream we run discovery across 8 regions, and the moment we introduced NATS JetStream as the spine between the crawler and the downstream consumers, the whole system got calmer. This post is the concrete version of how we did it: the stream config, the publishers, the consumers, and the mistakes we made that you can skip.
Why Not Just Use a Queue Table in SQLite
We already had SQLite everywhere, so the obvious move was a jobs table. We tried it. The problems showed up fast:
-
Polling latency vs. load tradeoff. Poll every second and you hammer the DB with mostly-empty
SELECTqueries across 8 regions. Poll every 30 seconds and your search index lags noticeably behind your crawler. -
No fan-out. One row, one worker. If the sitemap generator and the FTS5 indexer both need the same event, you either duplicate rows or invent a
consumed_bybitmask. Both are ugly. - Locking. SQLite's writer lock means the queue table and the actual data table start contending under the multi-region cron bursts we run.
- Cross-region delivery. Our regions aren't all on the same box. A queue table doesn't cross machines without you building a replication story on top.
JetStream solves all four: push-based delivery (no polling), multiple independent consumers off one stream, its own storage so it never touches your SQLite writer lock, and it's built to run as a cluster across regions. It's a single Go binary with no external dependencies, which matters a lot when your deploy story is FTP and a systemd unit rather than Kubernetes.
The Event Model
Before any code, we pinned down the events. The rule we follow: an event is a fact that already happened, named in the past tense, carrying enough data that a consumer never has to call back to ask "what did you mean."
Our subjects look like this:
-
video.discovered.<region>— crawler found a new video -
video.updated.<region>— metadata (title, view count) changed -
video.removed.<region>— video pulled or geo-blocked
The region in the subject is deliberate. It lets a consumer subscribe to video.discovered.us only, or video.discovered.* for everything, without us maintaining a routing table. Subject-based filtering is one of JetStream's best features and it's free.
Here's the stream, created once with the CLI. We keep this in a checked-in shell script so any region can bootstrap identically:
nats stream add VIDEO_EVENTS \
--subjects 'video.>' \
--storage file \
--retention limits \
--max-age 72h \
--max-bytes 2GB \
--discard old \
--dupe-window 2m \
--replicas 3
A few of these flags earned their place through pain:
-
--max-age 72hmeans a consumer can be down for up to three days and still catch up. Our FTP-deployed workers sometimes get restarted during a deploy window; 72 hours of buffer means a deploy is never a data-loss event. -
--dupe-window 2menables deduplication. If a crawler retries a publish with the sameNats-Msg-Id, JetStream drops the duplicate within a 2-minute window. This is the difference between "at-least-once" chaos and something that behaves like exactly-once for our purposes. -
--replicas 3keeps the stream alive if one region's node dies. If you're single-node, drop this to 1.
Publishing From PHP 8.4
Our crawler is PHP. There's no mature pure-PHP JetStream client that we trust, and honestly we didn't want a persistent connection living inside a short-lived cron process. So the publish path is deliberately dumb: PHP shells out to the nats CLI to publish. It's synchronous, it returns an exit code we can check, and it carries the dedup header.
This feels crude and it is exactly the right amount of crude for a cron-driven publisher:
<?php
declare(strict_types=1);
final class VideoEventPublisher
{
public function __construct(
private readonly string $region,
private readonly string $natsUrl = 'nats://127.0.0.1:4222',
) {}
/**
* Publish a discovery event. The message id is derived from the
* video id so a retried cron run is deduplicated by JetStream.
*/
public function discovered(string $videoId, array $payload): bool
{
$subject = sprintf('video.discovered.%s', $this->region);
$msgId = sprintf('disc-%s-%s', $this->region, $videoId);
$body = json_encode(
$payload + ['video_id' => $videoId, 'region' => $this->region],
JSON_THROW_ON_ERROR,
);
$cmd = sprintf(
'nats --server=%s pub %s %s -H %s 2>&1',
escapeshellarg($this->natsUrl),
escapeshellarg($subject),
escapeshellarg($body),
escapeshellarg('Nats-Msg-Id:' . $msgId),
);
exec($cmd, $output, $exitCode);
if ($exitCode !== 0) {
error_log(sprintf(
'[nats] publish failed subject=%s id=%s: %s',
$subject, $msgId, implode(' ', $output),
));
return false;
}
return true;
}
}
And the call site inside the fetch loop is now a single line that can fail without taking the crawl down with it:
$publisher = new VideoEventPublisher($region);
foreach ($newVideos as $video) {
$ok = $publisher->discovered($video['id'], [
'title' => $video['title'],
'channel_id' => $video['channel_id'],
'published_at' => $video['published_at'],
'thumbnail' => $video['thumbnail_url'],
]);
// The event is best-effort at publish time, but JetStream's 72h
// retention + dedup means a failed publish here is recoverable on
// the next cron pass. We never block the crawl on a downstream.
if (!$ok) {
$failedPublishes[] = $video['id'];
}
}
The important shift: the crawler's job is now "write the row and announce it." It has no opinion about what the search indexer or sitemap generator do with the announcement. Those consumers can be slow, restarted, or rewritten entirely and the crawler never notices.
Consuming In Go For The Search Indexer
The FTS5 indexer is the consumer that most wants to be fast and durable, so we wrote it in Go using the native JetStream client. This is a durable pull consumer — it has a name, JetStream remembers its position, and if it dies it resumes exactly where it left off.
package main
import (
"context"
"database/sql"
"encoding/json"
"log"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
_ "modernc.org/sqlite"
)
type VideoEvent struct {
VideoID string `json:"video_id"`
Region string `json:"region"`
Title string `json:"title"`
ChannelID string `json:"channel_id"`
PublishedAt string `json:"published_at"`
}
func main() {
nc, err := nats.Connect("nats://127.0.0.1:4222",
nats.MaxReconnects(-1),
nats.ReconnectWait(2*time.Second),
)
if err != nil {
log.Fatalf("connect: %v", err)
}
defer nc.Drain()
js, err := jetstream.New(nc)
if err != nil {
log.Fatalf("jetstream: %v", err)
}
db, err := sql.Open("sqlite", "/var/data/search_us.db")
if err != nil {
log.Fatalf("open db: %v", err)
}
defer db.Close()
ctx := context.Background()
// A durable consumer bound to one region's discovery subject.
// AckExplicit means we only advance after the FTS5 write succeeds.
cons, err := js.CreateOrUpdateConsumer(ctx, "VIDEO_EVENTS", jetstream.ConsumerConfig{
Durable: "fts5-indexer-us",
FilterSubject: "video.discovered.us",
AckPolicy: jetstream.AckExplicitPolicy,
MaxDeliver: 5,
AckWait: 30 * time.Second,
})
if err != nil {
log.Fatalf("consumer: %v", err)
}
_, err = cons.Consume(func(msg jetstream.Msg) {
var ev VideoEvent
if err := json.Unmarshal(msg.Data(), &ev); err != nil {
// Poison message: don't retry forever, park it.
log.Printf("bad json, terminating: %v", err)
msg.Term()
return
}
if err := indexVideo(db, ev); err != nil {
log.Printf("index %s failed, will redeliver: %v", ev.VideoID, err)
msg.Nak() // negative ack -> JetStream redelivers
return
}
if err := msg.Ack(); err != nil {
log.Printf("ack failed for %s: %v", ev.VideoID, err)
}
})
if err != nil {
log.Fatalf("consume: %v", err)
}
log.Println("fts5-indexer-us running")
select {} // block forever
}
func indexVideo(db *sql.DB, ev VideoEvent) error {
_, err := db.Exec(
`INSERT INTO video_fts (video_id, title, channel_id)
VALUES (?, ?, ?)
ON CONFLICT(video_id) DO UPDATE SET title = excluded.title`,
ev.VideoID, ev.Title, ev.ChannelID,
)
return err
}
The three message dispositions here are the whole game and worth spelling out:
-
msg.Ack()— the FTS5 write succeeded, advance the consumer. This only happens after SQLite confirms the row. -
msg.Nak()— a transient failure (SQLite was locked, disk was busy). JetStream redelivers, up toMaxDeliver: 5. -
msg.Term()— the message will never succeed (malformed JSON). Stop redelivering and move on. Without this, one bad payload blocks the consumer forever.
Because the consumer is durable and acks are explicit, an FTP deploy that restarts this binary is a non-event. It reconnects, JetStream replays anything unacked, and the index converges. That is the property we could never get from a polling loop.
The Sitemap Consumer In Python
Not every consumer needs Go's throughput. Our sitemap generator is a low-frequency Python service that just needs to know a URL changed so it can mark a region's sitemap dirty. Here it's fine to use a simpler push subscription, batching writes so we don't rewrite the sitemap on every single event:
import asyncio
import json
import nats
from nats.js.api import ConsumerConfig, AckPolicy
DIRTY_REGIONS: set[str] = set()
async def main() -> None:
nc = await nats.connect("nats://127.0.0.1:4222")
js = nc.jetstream()
async def handler(msg) -> None:
data = json.loads(msg.data)
DIRTY_REGIONS.add(data["region"])
await msg.ack()
# Subscribe to discovered + removed across all regions.
# A single durable consumer, filtered by a wildcard subject.
await js.subscribe(
"video.discovered.*",
durable="sitemap-marker",
cb=handler,
config=ConsumerConfig(ack_policy=AckPolicy.EXPLICIT),
)
await js.subscribe(
"video.removed.*",
durable="sitemap-marker-rm",
cb=handler,
config=ConsumerConfig(ack_policy=AckPolicy.EXPLICIT),
)
# Flush dirty regions every 5 minutes instead of per-event.
while True:
await asyncio.sleep(300)
for region in list(DIRTY_REGIONS):
regenerate_sitemap(region) # writes /public/sitemap_<region>.xml
DIRTY_REGIONS.discard(region)
def regenerate_sitemap(region: str) -> None:
print(f"[sitemap] regenerating region={region}")
# ... query SQLite, render XML, write file, ping IndexNow ...
if __name__ == "__main__":
asyncio.run(main())
The same stream feeds two completely different consumers with different languages, different cadences, and different failure tolerances. The FTS5 indexer cares about every message and acks precisely. The sitemap marker only cares that something changed in a region and batches accordingly. Neither knows the other exists. That decoupling is the entire point.
Things That Bit Us
A few hard-won lessons from running this in production across 8 regions:
-
Set a dedup window or regret it. Our crawler retries on API timeouts. Without
--dupe-window, a retried fetch republished every video and the FTS5 index got double-inserts until theON CONFLICTclause saved us. The dedup window catches it at the stream level so consumers never even see the duplicate. -
FilterSubjectbeats one giant consumer. Early on we had a single consumer readingvideo.>and switching on region in code. It worked but meant the US indexer was woken up for every European event only to discard it. Per-region filter subjects push the routing into JetStream, and each region's box only receives its own traffic. -
MaxDeliverwithoutTerm()is an infinite loop with extra steps. A malformed payload with onlyNak()and no terminal disposition will redeliver untilMaxDeliver, then dead-letter into a place you forgot to monitor. Always handle the poison-message case explicitly. -
Don't put JetStream on the same disk as your hot SQLite files. We initially pointed
--storage fileat the same volume as our search DBs. Under a multi-region cron burst, JetStream's fsyncs and SQLite's writes fought for IOPS. Separate volumes fixed the tail latency instantly. -
Monitor consumer lag, not just "is NATS up."
nats consumer report VIDEO_EVENTSshows unacked and pending counts per consumer. A consumer that's connected but falling behind is invisible to a simple health check. We alert on pending count crossing a threshold.
Was It Worth It
For a single-region, single-consumer setup, honestly, no — a queue table is simpler and you should use it. The moment we had multiple consumers wanting the same events, or events that needed to survive a deploy, or delivery that crossed machine boundaries, JetStream stopped being overkill and started being the obvious answer.
The concrete wins for us: the crawler's runtime dropped because it no longer blocks on downstream indexing, a deploy no longer risks losing in-flight work thanks to 72-hour retention, and adding a brand-new consumer — say, a real-time "trending now" ranker — is now a matter of writing a new durable consumer against the same stream, with zero changes to the publisher. The event is a fact that already exists; anyone who wants it can come and read it.
If you're running a discovery pipeline where one thing happening needs to notify several other things, and some of those things are allowed to be slow or occasionally down, put a durable log between them. NATS JetStream is a single binary, it speaks a subject hierarchy that maps cleanly onto regions, and it turns partial failures into recoverable ones. That trade — a bit more infrastructure for a lot less coupling — is one we'd make again.
Top comments (0)