Mogador: 13 million files through one workstation

Mogador is my media and text ingestion system.

That sentence is technically true and deeply rude to the amount of machinery involved. Mogador has processed more than 13 million files, keeps about 9.18 million documents searchable, and stores 11.77 million embedded text rows in ClickHouse. It started as “what if I just collected the data locally?” and quietly became actual infrastructure that a bunch of ML experiments now sit on, including ozut, my osu! beatmap generation project.

This post goes through the whole system: queues, storage, deduplication, models, concurrency, hardware, observability, the Rust rewrite, the osu! ingestion path, and what AWS would charge if I tried to host the same thing and handed Jeff Bezos my debit card as a little enrichment toy.

I’m leaving most upstream data sources unnamed on purpose. This post is about system design, not a directory of sources. The osu! path is the exception: it’s a friendly integration built on public beatmap metadata and downloads, and it feeds ozut directly.

What a scraping pipeline is made of

Collecting a file is never one step. Something has to find it, download it, check it isn’t a copy of something you already have, convert it to a sane format, tag it, embed it, and make it searchable. Each of those steps is slow in its own way, and each fails in its own way.

So Mogador splits them into separate programs, called workers, connected by message queues. A worker takes a message off one queue, does its one job, and puts a new message on the next queue. If a worker crashes, its message goes back on the queue and another worker picks it up.

A message is a small JSON document describing the work. Here’s the shape of a download job (the values are made up):

{
  "source": "source-a",
  "source_id": 123456,
  "post_id": "e9c86a66-597c-42da-86ed-977af1438195",
  "media_url": "https://cdn.example.org/a/123456.mp4",
  "view_url": "https://example.org/post/123456",
  "referer": "https://example.org/",
  "tags": ["tag_a", "tag_b"]
}

It says: go fetch this file for post 123456 of source A, send this referer, remember these tags. The downloader doesn’t need to know anything else.

Where it stands

Measured on formidable, my workstation, on 2026-04-29:

Component	Snapshot
Searchable documents (Elasticsearch posts)	9,183,737
Deleted docs waiting for a merge	496,766
Elasticsearch primary store	330.7 GB
Tags index documents	231,226
ClickHouse lake.docs rows	11,767,862
ClickHouse lake.embeddings rows	11,768,211
BGE-M3 vectors, raw size	44.89 GiB
ClickHouse embedding table on disk	34.20 GiB
ClickHouse document table on disk	824.98 MiB
Media disk	6.47 TiB used of 7.28 TiB
MinIO staging disk	100 GB used

The flow, end to end:

1. Scrapers discover records.
2. Each record is inserted or claimed exactly once.
3. Download jobs go into RabbitMQ.
4. The downloader streams the media into object storage.
5. The optimizer normalizes the media and deduplicates it.
6. The tagger, embedder, thumbnailer and indexer enrich it.
7. Elasticsearch serves search and retrieval.
8. ClickHouse keeps the analytical text lake.
9. Kafka receives structured logs for audit and replay.

The machine: formidable

Mogador runs on formidable, the same machine as in Everything I Self-Host at Home.

Resource	Value
CPU	AMD Ryzen 9 7950X, 16 cores / 32 threads
RAM	61 GiB
GPU	NVIDIA GeForce RTX 4090, 24 GiB VRAM
Driver	595.58.03
OS / root	encrypted btrfs on a 1 TB NVMe
Fast work disk	1.8 TiB SSD (javelin)
Media disks	RAID1 of two 8 TB disks, ~7.28 TiB usable
Media usage	6.47 TiB used, 0.81 TiB free, 88.9% full

The one GPU is shared by the whole pipeline. The same RTX 4090 runs image tagging, text embeddings, thumbnail and transcode work, comment generation experiments, and whatever else I’m training that week. So every GPU worker sits behind a queue with conservative concurrency, instead of every process sprinting into CUDA like a toddler into traffic.

One job per database

Layer	Its job
MinIO	S3-compatible staging for raw and intermediate files
Local filesystem	final media and thumbnails
PostgreSQL	scraper state, claims, old post metadata
Elasticsearch	search, filters, ranking, vector retrieval
ClickHouse	text lake, embeddings, analytics, feature work
RocksDB	deduplication state and publish reservations
Kafka	event log for audit, replay, downstream consumers
RabbitMQ	task queues, with quorum durability
etcd	live config and per-service overrides

It sounds like a lot until you read the jobs as questions. RabbitMQ decides what should happen next. Kafka records what happened. PostgreSQL answers have we seen this scraper item before? RocksDB answers is this media a duplicate, or already being processed? Elasticsearch answers can the app search this fast? ClickHouse answers can I aggregate this without crying?

Databases are happiest when they aren’t forced to do each other’s hobbies.

The queues

Queue	What goes in it
to_download	fetch original media or external files
to_optimize	normalize, transcode, deduplicate
to_tag	tag with the vision model
to_embed	embed media search documents
to_embed_text	embed standalone text and comments into ClickHouse
to_thumbnail	generate previews
to_index	write final documents to Elasticsearch
to_hash	backfill hash and dedupe state
to_comment	generate synthetic comments for experiments

Every queue is a RabbitMQ quorum queue: its contents are replicated through a consensus protocol instead of living in one place. That costs a bit more than a classic queue, but a worker crash becomes a retry instead of lost data.

The Rust helper that declares one:

pub async fn declare_quorum_queue(&self, name: &str)
    -> BoxResult<()>
{
    let mut arguments = FieldTable::default();
    arguments.insert(
        ShortString::from("x-queue-type"),
        AMQPValue::LongString(LongString::from("quorum")),
    );

    self.publish_channel
        .queue_declare(
            name.into(),
            QueueDeclareOptions {
                durable: true,
                auto_delete: false,
                exclusive: false,
                passive: false,
                nowait: false,
            },
            arguments,
        )
        .await?;

    Ok(())
}

The contract every worker follows: durable queues, JSON payloads, persistent delivery (delivery mode 2), message IDs, and a clear decision on failure between retrying and dead-lettering.

Why I rewrote it in Rust

The first pipeline was mostly Python. That got the system moving, but long-running workers need predictable memory, clean cancellation, backpressure and explicit concurrency, and Python made all four hard.

The rewrite moved the hot path to Rust:

Service	Runtime	Role
scraper kernel	Rust	shared traits: adapters, checkpoints, claims, publishing
downloader	Rust	stream downloads into object storage, retry, dead-letter
optimizer	Rust	transcode, hash, dedupe, publish
thumbnailer	Rust	preview images and videos
indexer	Rust	batched Elasticsearch writes, final media moves
source adapters	Rust	per-source discovery workers
embedding	Python	BGE-M3 on the GPU, ClickHouse/Postgres writers
tagger	Python	GPU image and video tagger on ONNX Runtime
commenter	Python	local LLM comment generation experiments

The Rust services share a kernel built around a few traits:

pub trait NativeAdapter: Send + Sync {
    type Item;
    type Checkpoint: Checkpoint;
    type Record;
    type Message;

    fn item_key(&self, item: &Self::Item) -> String;
    fn checkpoint(&self, item: &Self::Item)
        -> Self::Checkpoint;
    fn classify(&self, item: Self::Item)
        -> BoxResult<NativeAction<Self::Record, Self::Message>>;
}

This is the main reason adding a source got fast. An adapter only answers five questions:

1. How do I page through upstream records?
2. What’s the stable key of an item?
3. Which checkpoint only ever moves forward?
4. Should this item be skipped, stored, or stored and published?
5. Which queue message comes out of it?

Everything else is shared: claims, checkpoints, duplicates, publishing, retries, logging and metrics.

Claiming every item

Scraping systems mostly die from duplicate work, partial work, and weird half-finished states. Mogador makes every item idempotent: processing it twice has the same effect as processing it once. The way it gets there is a claim. Before touching an item, a worker has to claim its key:

match completion_store.claim(&key).await? {
    ClaimOutcome::Completed => {
        report.duplicates += 1;
        advance_checkpoint(
            checkpoint_store,
            checkpoint_key,
            current_checkpoint,
            checkpoint,
        )
        .await?;
        continue;
    }
    ClaimOutcome::Busy => {
        return Err(KernelError::Contention(
            "item is already being processed by another worker",
        ));
    }
    ClaimOutcome::Claimed => {}
}

Claim result	Meaning
completed	already handled: skip it, advance the checkpoint if safe
busy	another worker owns it: stop before causing duplicate side effects
claimed	this worker may go ahead

In practice, workers can restart, scrapers can overlap, pages can repeat, and the same media still ends up stored once.

Downloading: stream, upload, publish

downloader-rs takes jobs from to_download, checks them against policy, streams the file into MinIO, and puts a job on to_optimize.

It streams and counts bytes instead of loading whole files into memory. It rejects blacklisted extensions, enforces a maximum size, sends per-source cookies when configured, and treats a 404 as a dead-letter event instead of eternal retry confetti.

let downloaded = self
    .http
    .stream(
        &payload.media_url,
        payload.referer.as_deref(),
        &cookies,
        max_bytes,
        &self.metrics,
    )
    .await?;

let s3_key =
    format!("{}/{}{}", payload.source, payload.post_id, ext);

self.storage
    .upload_stream(
        &s3_key,
        downloaded.body_stream,
        downloaded.content_length,
        downloaded.mime.as_deref(),
    )
    .await?;

self.amqp
    .publish_json(&self.queue_to_optimize, &optimize_payload,
        None, None)
    .await?;

That’s deliberately boring: stream, upload, publish, acknowledge. On failure, classify why, then retry or dead-letter.

The optimizer: dedupe and normalize

The optimizer is the busiest and most important worker. For each staged file, it downloads it into a temp directory, hashes it, reserves a dedupe record, transcodes it, uploads the result, and publishes the next jobs.

It computes two kinds of hash:

• An exact hash, BLAKE3. Change one byte and the hash changes completely. It catches byte-identical copies.
• Perceptual hashes (aHash, dHash, pHash). They describe what the image looks like, so a re-encoded, resized or brightened copy gets the same (or a very close) hash.

For images, the optimizer computes BLAKE3 and the three perceptual hashes, then normalizes them with libvips into WebP/AVIF-friendly versions. For videos, it computes BLAKE3, samples frames, hashes each sampled frame perceptually, and normalizes with ffmpeg, including AV1 encodes where the storage and quality trade-off is worth it.

The dedupe state lives in RocksDB, split into column families (separate key spaces inside one database) for exact and perceptual matches:

const CF_EXACT_PENDING: &str = "exact_pending";
const CF_EXACT_COMMITTED: &str = "exact_committed";
const CF_IMAGE_PENDING: &str = "image_pending";
const CF_IMAGE_COMMITTED: &str = "image_committed";
const CF_VIDEO_PENDING: &str = "video_pending";
const CF_VIDEO_COMMITTED: &str = "video_committed";

Each media object moves through Pending, ReadyToPublish, Publishing, Published or Aborted. Dedupe and publishing have to share this state machine: if two workers get the same media, the second can reuse the first one’s prepared result instead of transcoding and uploading it again.

The optimizer times each stage separately:

Stage	What it measures
download	fetching the staged file from MinIO
hash	exact and perceptual hashing
dedupe	reserving or detecting a duplicate in RocksDB
transcode	image and video normalization
upload	writing the normalized output
publish	sending the next messages

When things slow down, that tells me whether the villain is disk IO, ffmpeg’s AV1 work, libvips, a saturated GPU, MinIO, RocksDB, or a queue that has started making little haunted noises.

Storing millions of files: Git-style fan-out

Millions of files in one directory make every tool slow. Final media and thumbnails are spread out the way Git stores its objects: the first two characters of the name become a directory, the next two a subdirectory.

media/
├── e9/
│   ├── c8/
│   │   └── e9c86a66-597c-42da-86ed-977af1438195.mp4
│   └── d1/
│       └── e9d1...
└── f0/
    └── ...

With hex names, that’s 256 × 256 = 65,536 directories, so each one stays small enough for normal filesystem tools. The helper:

fn fanout_path(root: &Path, id: Uuid, extension: &str)
    -> PathBuf
{
    let name = id.hyphenated().to_string();
    let a = &name[0..2];
    let b = &name[2..4];
    root.join(a).join(b).join(format!("{name}.{extension}"))
}

Tagging on the 4090

The tagger is Python because the model ecosystem is Python-shaped, and I have chosen peace in this tiny corner of the universe.

Setting	Value
Model	SmilingWolf/wd-eva02-large-tagger-v3
Runtime	ONNX Runtime, GPU
Input size	448 × 448
Video sampling	0.5 frames per second
Batch size	8
General tag threshold	0.35
Character tag threshold	0.75

It’s a WaifuDiffusion-family tagger: for each image or video frame, it outputs ranked tags for content, characters, style and rough safety categories. Those feed search filters, dataset exports and later ML training. The two thresholds are the minimum confidence for a tag to be kept, and the bar is higher for characters than for general tags.

The worker consumes to_tag, pulls the media, extracts frames, runs batched inference, writes the tags, and publishes to indexing. It exposes Prometheus metrics for model loading, batch size, latency, tag counts and dead letters, plus a few HTTP admin endpoints.

The model loads on first use and is dropped again when its 600-second TTL runs out, so its GPU memory goes back to the rest of the machine.

The interesting part is the batching. Messages arrive one at a time, but the GPU is much more efficient on a batch. So the worker waits a very short time to collect more:

MODEL_REPO_ID = "SmilingWolf/wd-eva02-large-tagger-v3"
IMAGE_SIZE = 448
EXTRACT_FPS = 0.5
BATCH_SIZE = 8
INFERENCE_BATCH_WAIT_MS = 8.0
GENERAL_THRESHOLD = 0.35
CHARACTER_THRESHOLD = 0.75
DEFAULT_MODEL_TTL_SECONDS = 600

Eight milliseconds sounds silly until it saves you from launching one GPU inference per message like a tiny CUDA woodpecker.

Embeddings: BGE-M3 into ClickHouse

An embedding turns a text into a vector of numbers so that similar meanings land close together. It’s what makes “find comments like this one” possible.

The embedding service reads two queues:

Queue	What it embeds
to_embed	media search documents: metadata, tags, captions and comments when present, source fields
to_embed_text	standalone text and comments, into ClickHouse

Setting	Value
Model	BAAI/bge-m3
Library	FlagEmbedding
Dimensions	1024
Rows in ClickHouse	11,768,211
Raw vector size	44.89 GiB
Table size on disk, compressed	34.20 GiB

The ClickHouse tables:

CREATE TABLE IF NOT EXISTS lake.docs
(
  source     LowCardinality(String),
  source_id  String,
  doc_uuid   UUID,
  updated_at DateTime64(3),
  ingest_at  DateTime64(3),
  author     String,
  text       String CODEC(ZSTD(3)),
  meta_json  String CODEC(ZSTD(3))
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(ingest_at)
ORDER BY (source, source_id, updated_at);

CREATE TABLE IF NOT EXISTS lake.embeddings
(
  model      LowCardinality(String),
  doc_uuid   UUID,
  updated_at DateTime64(3),
  embedding  Array(Float32)
)
ENGINE = MergeTree
PARTITION BY (model, toYYYYMM(updated_at))
ORDER BY (model, doc_uuid);

What’s in the text lake:

Source class	Rows	Characters
media-comment corpus A	5,263,380	394,130,823
media-comment corpus B	2,401,321	306,184,712
chat/comment corpus	1,306,044	37,656,367
smaller corpora	81,457	5,772,478

Source names left out again; the numbers are the interesting part.

The embedder reports metrics for per-pipeline outcomes, end-to-end and per-stage latency, cache hits and misses, batch size, why each batch was flushed, writer queue depth, rows inserted into ClickHouse, and model loads.

The lesson from running BGE-M3 locally: embeddings are only cheap if you batch. One 4090 gets through a lot of text, but if every message calls model.encode() on its own, the GPU spends most of its time on overhead. Mogador batches inference according to queue pressure, and batches its ClickHouse writes too.

Elasticsearch: 9 million searchable documents

Index	Documents	Size
posts	9,183,737	330.7 GB
tags	231,226	32.4 MB

The posts index holds dense vectors alongside the metadata. The cluster currently reports 45 indices, 231 segments, and vector fields of 1024 and 2560 dimensions.

The indexer is Rust. Between the queue and Elasticsearch sits a batcher, and every batch goes through the bulk API, one request for many documents:

fn should_flush(
    batch: &[PreparedMessage],
    batch_bytes: usize,
    config: &AppConfig,
) -> bool {
    batch.len() >= config.batch_size
        || batch_bytes >= config.batch_max_bytes
}

async fn flush_batch(
    batch: &mut Vec<PreparedMessage>,
    batch_bytes: &mut usize,
    ctx: &BatcherContext,
) {
    if batch.is_empty() {
        *batch_bytes = 0;
        return;
    }

    match ctx.es.bulk_index(batch).await {
        Ok(results) => {
            for (item, result) in batch.drain(..).zip(results) {
                match result {
                    Ok(()) => complete_success(item, ctx).await,
                    Err(error) => {
                        retry_prepared(item, ctx, &error).await
                    }
                }
            }
        }
        Err(error) => {
            let reason = error.to_string();
            for item in batch.drain(..) {
                retry_prepared(item, ctx, &reason).await;
            }
        }
    }

    *batch_bytes = 0;
}

A failed bulk request retries every item in it. A partly failed one retries only the items that failed. Documents are prepared concurrently, then written in bulk instead of one request each, which keeps throughput predictable.

Kafka: the audit log

RabbitMQ is for scheduling work. Kafka is for facts. Every service writes compact JSON events to Kafka topics, grouped by domain:

Topic	Example events
mogador.scraper.events	page_fetched, item_claimed, item_skipped, checkpoint_advanced
mogador.download.events	download_started, download_completed, download_failed, dead_lettered
mogador.optimize.events	hash_computed, duplicate_detected, transcode_completed, publish_ready
mogador.ml.events	tagger_loaded, embedding_batch_flushed, model_evicted, inference_failed
mogador.index.events	bulk_index_started, bulk_index_partial, bulk_index_completed
mogador.storage.events	object_written, object_deleted, fanout_path_assigned

An event looks like this:

{
  "event_id": "01JZ...",
  "trace_id": "optimizer:queued-posts/example-key",
  "service": "optimizer-rs",
  "event": "duplicate_detected",
  "source_class": "image-board",
  "post_id": "9f0b9e6d-...",
  "media_kind": "image",
  "match_kind": "perceptual",
  "duration_ms": 12.7,
  "queue": "to_optimize",
  "attempt": 0,
  "timestamp": "2026-04-29T21:17:00.000Z"
}

The optimizer found a perceptual duplicate of an image, on the first attempt, in 12.7 ms.

This gives Mogador a history I can replay, without making Elasticsearch or ClickHouse the source of truth for events. If a file disappeared, got duplicated, failed tagging, or was indexed with stale metadata, I can rebuild its timeline. “I swear the worker did something weird yesterday” makes a terrible observability strategy.

Adding a source

A new source adapter needs:

1. a config file,
2. a worker implementing the kernel traits,
3. a stable item key,
4. a checkpoint strategy,
5. a classifier that produces a DownloadMessage or a TextEmbedMessage,
6. an entry in the compose file.

The source-specific code stops at discovery. Tagging, embedding, transcoding, dedupe, thumbnails and indexing belong to the shared pipeline. The adapter finds records and publishes work, using the shared message types:

pub use model::{
    DownloadMessage,
    MogadorPostRecord,
    OptimizeMessage,
    TextEmbedMessage,
    stable_post_id,
};

New formats: transformers

The optimizer treats every format as something to transform. A source emits a generic download job, and the pipeline decides how to normalize it:

Input	Output
still image	WebP
animation or video	MP4 preview or normalized MP4
text or comment	ClickHouse row + BGE-M3 embedding
osu! beatmap archive	cleaned archive + Opus audio + patched .osu files

Each transformer implements one contract:

trait Transformer {
    type Input;
    type Output;

    fn inspect(&self, input: &Self::Input)
        -> Result<MediaKind>;
    fn transform(&self, input: &Self::Input, output: &Path)
        -> Result<Self::Output>;
    fn downstream_messages(&self, output: &Self::Output)
        -> Result<Vec<QueueMessage>>;
}

Supporting a new format never means rewriting a scraper. The scraper emits a file; the pipeline either has a transformer for it or dead-letters it with a reason.

osu! beatmaps for ozut

The osu! path is the friendliest scraper in the system. It feeds ozut, which generates osu! beatmaps. It has two parts:

1. syncing metadata from osu!‘s beatmapset search,
2. pushing the beatmap archives through the Mogador pipeline.

The sync keeps a local SQLite database of beatmapsets, beatmaps, cursor state, download status and hashes, and has two modes:

Mode	What it does
cursor sync	walks every ranked beatmapset, page by page
incremental sync	fetches until it reaches the newest set it already has

It’s defensive about the pagination cursor, because a cursor that doesn’t move means an infinite loop:

if page.cursor_string and page.cursor_string == previous_cursor:
    raise RuntimeError(
        "cursor did not advance, aborting to avoid an infinite loop"
    )

Once stored, archives go straight onto to_download and behave like any other file, with their own Rust transformer. The transformer:

1. extracts the .osz archive,
2. converts the main audio track to Opus at 128 kbps,
3. rewrites every .osu file so its AudioFilename: points at the new Opus file,
4. drops everything training doesn’t need: custom hitsounds, storyboards, videos, backgrounds.

The patcher:

fn patch_audio_filename(osu_text: &str, opus_name: &str)
    -> String
{
    osu_text
        .lines()
        .map(|line| {
            if line.starts_with("AudioFilename:") {
                format!("AudioFilename: {opus_name}")
            } else {
                line.to_owned()
            }
        })
        .collect::<Vec<_>>()
        .join("\n")
}

What’s left is what ozut learns from: timing, hit objects, difficulty metadata and audio, with every .osu file pointing at audio.opus. The corpus is over 200 GB of maps, 301,374 beatmaps, and it goes from public metadata to model-ready files through the same queues as everything else.

What Mogador feeds

Comment styles and topics

The comment corpus fed a DeBERTa-based style classifier (the comment generator post tells that story), and the embeddings made it practical to cluster comments and model their topics.

Comments are stored twice: as text and metadata in ClickHouse, and as BGE-M3 vectors in lake.embeddings. So I can ask analytical and retrieval questions of the same data:

• Which sources produce which mix of styles?
• Which comments cluster together by meaning?
• Which topics go with which tag families?
• Which styles are worth generating synthetically?

LLM fine-tuning

The same corpus supports local fine-tuning with Unsloth. The commenter service runs a local language model with LoRA checkpoints to generate comments for a given style and set of tags.

Setting	Value
Base model	HuggingFaceTB/SmolLM3-3B
Adapter	LoRA checkpoint
Runtime	PyTorch + PEFT + Transformers
GPU dtype	bfloat16 on CUDA
Queue	to_comment

It stays out of the public search path on purpose. It’s an experiment for controlled data work.

Image LoRAs

The media side has built image datasets for SDXL and SD3 LoRA experiments. The tags and the search index let me select a dataset by visual concept, style tags, dimensions and dedupe status, then export a training manifest. Instead of hand-picking files one at a time, I write a query.

ozut

ozut trains on the osu! side. Cleaned archives, converted audio and stripped assets make it practical to train on hundreds of gigabytes of maps without dragging unrelated files through every experiment.

Recommendations and reranking

Mogador is also the dataset behind my recommendation and reranking experiments:

• vector retrieval with BGE-M3,
• tag-based candidate generation,
• Elasticsearch ranking features,
• ClickHouse analytics for exposure, click and favorite events,
• reranking candidate logs,
• local evaluation loops.

There’s already a separate set of ClickHouse tables for user events, exposures, favorites, list impressions and ranking candidates. The scraped corpus became a retrieval playground because it has what recommender systems always want and rarely get cheaply: lots of structured items with tags, vectors and metadata.

Concurrency, per service

Every service gets the concurrency shape it deserves, and they’re deliberately different:

Service	Concurrency
Rust scrapers	periodic batches, in checkpoint order
downloader	async streaming, bounded prefetch
optimizer	queue consumer + blocking hash/transcode tasks + RocksDB reservations
tagger	async queue + GPU microbatching
embedder	async queues + microbatched inference + batched DB writes
thumbnailer	one temp dir per message + blocking transcode tasks
indexer	concurrent preparation + one bulk batcher

The indexer shows the pattern best: workers prepare items concurrently, and a single batcher flushes them to Elasticsearch by count, estimated size, or timeout. That avoids both extremes, one request per message and unbounded memory growth.

The optimizer runs CPU-heavy hashing and transcoding with spawn_blocking, so the async runtime stays responsive while the 7950X does real work.

The GPU services use tiny wait windows to build batches. The goal is to avoid the saddest batch size, 1, forever, while keeping latency low.

Retries and dead letters

Retries are explicit and counted. Every failure gets a reason label:

• http_404
• unsupported_media
• max_retries_exceeded
• pending_duplicate
• publish_error
• clickhouse_error
• db_error
• model_error
• invalid_payload
• missing_field

A dead-letter queue is a labeled parking lot. One full of unlabeled payloads is a blind spot.

Observability

Three layers:

1. Prometheus metrics from the services and exporters.
2. The Kafka event log.
3. Each store’s own diagnostics: RabbitMQ, Elasticsearch, ClickHouse, PostgreSQL.

The compose stack runs a RabbitMQ exporter, an Elasticsearch exporter, node-exporter, cAdvisor, Prometheus with 30 days of retention, Grafana dashboards, and a custom Mogador metrics scraper.

The custom scraper reads Elasticsearch index stats and emits corpus-level gauges. Generic exporters are good at telling you a database exists. They’re less good at answering “is my corpus growing the way I expect, or has it become a mushroom?”

Why both ClickHouse and Elasticsearch

Elasticsearch is for online search: full-text-ish search, filter-heavy queries, vector retrieval, document lookup, ranking experiments.

ClickHouse is for analytical scans: counting millions of rows fast, compressing text columns, keeping embedding history, grouping by source, model or month, offline analysis.

Making one of them do both jobs is possible the way eating soup with a fork is possible. You can prove your dedication; you can’t prove your judgment.

What it costs at home

Mostly electricity and disks. formidable is my workstation and exists anyway, so Mogador’s extra cost is far below any cloud equivalent.

Mount	Capacity	Used	Free	Usage
media RAID1	7.28 TiB	6.47 TiB	0.81 TiB	88.9%
fast SSD (javelin)	1.82 TiB	1.30 TiB	0.52 TiB	71.5%
MinIO staging SSD	119 GiB	99 GiB	20 GiB	83.1%

The two 8 TB disks in RAID1 give roughly one disk of usable space, mirrored. That protects against one dead disk and keeps the corpus online. Backups are separate.

What it would cost on AWS

A rough estimate, not a quote. AWS prices depend on region, commitments, storage class, egress, and managed-service minimums. I used the public pricing (S3, SQS, EBS, OpenSearch, ClickHouse Cloud, MediaConvert) and these assumptions:

• 9.46 TiB of objects in S3 Standard: the 7.28 TiB media disk plus 30% room to grow, since disks don’t stop filling themselves,
• two 8 TB disks’ worth of gp3 block storage, for RAID1-like durability,
• 13M processed files, each going through 6 queue hops,
• one send, receive and delete per hop,
• OpenSearch with production-style redundancy,
• ClickHouse Cloud or an equivalent managed tier,
• one CPU instance comparable to the 7950X,
• one 24 GB GPU instance comparable to the 4090, for the ML services,
• MediaConvert priced separately, as a one-off or monthly cost depending on workload.

Component	Estimate
S3 Standard, 9.46 TiB	~$223/mo
S3 requests for 13M objects	~$211 per full processing pass
SQS, ~234M billed requests	~$93 per full processing pass
gp3, 16 TB raw (RAID1-equivalent)	~$1,311/mo
OpenSearch compute	~$366–$731/mo
OpenSearch storage, 3 × 331 GB	~$121/mo
ClickHouse Cloud / managed tier	~$390–$1,560/mo
CPU instance	~$894–$1,340/mo
24 GB GPU instance	~$734/mo
1 TB internet egress	~$92/mo

Scenario	Monthly total
no egress	~$4,300–$6,300
1 TB egress	~$4,400–$6,400
5 TB egress	~$4,800–$6,800

Then MediaConvert, the jump scare. It bills per output minute, with multipliers for resolution and quality. One pass over 1.3 million videos lands anywhere from ~$4,875 for very short, simple outputs to ~$78,000 for longer, higher-quality ones.

At that point the cloud version is mostly a budgeting exercise: every queue hop, transcode minute, vector table, replica and egress path becomes a line item with sharp little teeth.

The rules it runs on

1. Every item has a stable key.
2. Every worker can retry safely.
3. Every expensive stage sits behind a queue.
4. Every storage layer has one main job.
5. Deduplication happens before long-term storage.
6. GPU inference is batched.
7. Bulk writes are batched.
8. Config lives in one place.
9. Logs are structured.
10. A new source is a small adapter, not a new platform.