DEV Community: Umang Aggarwal

Building an LLM Inference Simulator

Umang Aggarwal — Tue, 10 Nov 2026 14:02:13 +0000

I built a small LLM server that has no model and no GPU, but slows down under load like a real one does.

Most of what we build around LLMs sits in front of the model: gateways, routers, rate limiters, streaming clients and observability dashboards. Testing those layers against real GPUs is expensive. Testing it against a mock is cheap, but a mock either answers instantly at any load or adds a fixed delay or random jitter. It never queues, never runs out of KV cache and never preempts a request halfway through. So the failure modes we need to engineer never show up.

I wanted something in between: an OpenAI-compatible server that is deterministic, runs on a laptop or in CI, and whose time to first token (TTFT) and inter-token latency (ITL) respond to load like a real serving engine’s.

I followed the rule of predicting numbers before running load tests, then measuring the diversion and fixing the fidelity issues. The source code is shared at the end of this article.

Scope: What it is, and what it isn’t

It answers one question: how does my system behave when the model slows down? It does not predict how fast a particular model runs on particular GPUs. Only the real engine, or a simulator built from profiled kernels, can answer that.

A simulator is only useful if you know where to stop trusting it. Therefore, I would like to clarify the fidelity before we go through the architecture.

Architecture: what is said vs when it is sent

A real LLM server answers two things at once: what the model says, and when each token arrives. The simulator splits them.

The generator (llm/) decides what is said. It counts prompt tokens, matches the last user message against scripted responses in responses.yml, picks a variant deterministically and truncates to max_tokens. It never deals with time.
The engine (simulator/) decides when each token is sent. It takes the generator's plan (prompt length, output tokens) and the request's arrival time, and yields one TokenEvent per token with its simulated time. It never deals with content.
The two share only types.py , which imports nothing from the project, so neither layer can reach into the other.
runtime.py is the composition root. It is the only module that reads the parsed config, and it builds one generator and one engine per model at startup. A broken config stops the server before it accepts traffic.
Engines report metrics through callbacks wired in runtime.py, so the simulator never imports the metrics layer.

The split is what made the project tractable. The content side stayed a simple YAML lookup while the timing side grew from fixed delays to a full scheduler, without either side changing the other.

The generator orchestrates resulting matcher and tokenizer:

def generate(self, request: GenerationRequest) -> GenerationPlan:
        raw = _last_user_message(request.messages)
        # Matching and selection ignore surrounding whitespace, so "hello\n" and
        # "hello" get the same answer; {last_user_message} keeps the raw text.
        text = raw.strip()

        prompt_tokens = self._tokenizer.count_messages(request.messages)
        # Checked before matching, so rejected requests cost nothing more.
        limit = self._completion_limit(prompt_tokens, request.max_tokens)

        result = self._matcher.match(text)
        seed = self._seed if request.seed is None else request.seed
        digest = _digest(seed, result.entry_id, text)

        choices = result.body.choices
        index = _hash_int(digest, 0) % len(choices)
        variant = choices[index]
        variables = {
            **result.groups,
            "last_user_message": raw,
            "model": request.model,
        }
        tokens = self._tokenizer.encode(_render(variant.parts, variables))
        return GenerationPlan(
            prompt_tokens=prompt_tokens,
            tokens=tuple(tokens),
            finish_reason="length" if truncated else "stop",
            entry_id=result.entry_id,
            variant=index,
        )

The matcher matches the incoming prompt with a deterministic response configuration. The response configuration follows a custom DSL:

version: 1
seed: 42 # default seed for choosing variants

responses: # checked top to bottom; the first match wins
  - id: greeting
    match: { exact: ["hi", "hello"] }
    content: "Hello! How can I help you today?"

  - id: weather
    match: { regex: "weather in (?P<city>\\w+)" }
    variants: # one is chosen per request, deterministically
      - "It's sunny in {city} today."
      - "Expect light rain in {city}."

  - id: echo
    match: { starts_with: "/echo" }
    content: "You said: {last_user_message}"

  - id: long-answer
    match: { contains: "explain" }
    synthetic: { tokens: [200, 400] } # filler text of 200 to 400 tokens

default: # required: used when nothing matches
  variants:
    - "I'm a simulated model ({model})."
    - "This is a canned response."

The tokenizer is a regex implementation that enables processing the pattern:

_PATTERN = re.compile(
    r"""
    '(?:[sdmt]|ll|ve|re) # contractions: 's 'd 'm 't 'll 've 're
    | []?[%(cjk)s] # CJK, kana and Hangul, one character per token
    | []?(?:(?![%(cjk)s])[^\W\d_])+ # words (other letters), with an optional leading space
    | []?\d{1,3} # numbers, in groups of up to three digits
    | []?[^\s\w]+ # runs of punctuation and symbols
    | \s+(?!\S) # whitespace, leaving the last space to start the next token
    | \s+ # whitespace left over, such as a newline before a word
    | [\s\S] # anything else (e.g. underscores), so encoding is lossless
    """ % {"cjk": _CJK},
    re.IGNORECASE | re.VERBOSE,
)

def encode(self, text: str) -> List[str]:
    return self._PATTERN.findall(text)

def count(self, text: str) -> int:
    return sum(1 for _ in self._PATTERN.finditer(text))

There are three engines behind one interface:

The engine is chosen per model in models.yml, so one server can serve a fixed-rate model and a batched one side by side. The configuration looks like:

hardware:
  h100-80gb:
    hbm_gib: 80
    memory_bandwidth_gbps: 3000
    peak_tflops: 989

models:
  - name: qwen-mock-batched
    context_length: 32768
    tokenizer: regex
    engine: scheduler
    hardware: h100-80gb
    weights:
      parameters_billions: 6.74
      bytes_per_parameter: 2
    kv_cache:
      bytes_per_token: 524288
      block_size: 16
      gpu_memory_utilization: 0.9
      watermark: 0.01
    scheduler:
      max_batch_tokens: 8192
      max_num_sequences: 64
    latency:
      step_overhead_ms: 0
    efficiency:
      compute: 1.0
      memory: 1.0

The engine: one interface, two very different timelines

Every engine does one job: turn a plan into a timed stream of tokens. The interface is small on purpose:

class Engine(ABC):
    """Turns a plan into a timed stream of token events.
    Engines never change what is said, only when."""

    name: ClassVar[str]
    # models.yml sections a model must define to use this engine; checked at startup
    required_sections: ClassVar[Tuple[str, ...]] = ()

    @abstractmethod
    def run(self, plan: GenerationPlan, arrival: float) -> AsyncIterator[TokenEvent]:
        """Yield the plan's tokens in order; deadlines are computed from arrival."""

    async def start(self) -> None: ... # e.g. launch a background step loop
    async def stop(self) -> None: ...

The two types that cross the boundary carry no HTTP and no config:

GenerationPlan comes from the generator: prompt_tokens, the output tokens, finish_reason, and which response entry answered.
TokenEvent goes back to the SSE layer: index, text, sim_time (simulated seconds after arrival when the token is due) and lag (how late it was actually released).

The API layer pulls tokens with async for event in engine.run(plan, arrival) and writes a chunk for each. It doesn't know or care which engine is behind it. When a client disconnects, the API closes the iterator, and the engine's finally: block frees whatever the request held.

Engines take plain parameters, never config classes. runtime.py reads models.yml, checks required_sections, and builds each engine with numbers. That keeps every engine testable with a fake clock and a few arguments.

Fixed-rate: each request owns its timeline

The fixed-rate engine has no shared state. A request’s whole timeline is known the moment it arrives:

def timeline(self, plan: GenerationPlan) -> Tuple[float, ...]:
    """Simulated seconds after arrival at which each token is due."""
    ttft = self._overhead + plan.prompt_tokens / self._prefill_rate
    return tuple(ttft + index * self._itl for index in range(len(plan.tokens)))

async def run(self, plan, arrival):
    for index, (text, sim_time) in enumerate(zip(plan.tokens, self.timeline(plan))):
        lag = await sleep_until(arrival + sim_time, self._clock, self._sleep)
        yield TokenEvent(index, text, sim_time, lag)

That is the whole engine. A thousand concurrent requests means a thousand independent coroutines, each sleeping to its own deadlines. Load changes nothing about the simulated timeline.

Scheduler: one step loop owns the GPU

The scheduler inverts this. No request knows its timeline in advance, because when its tokens come out depends on who else is in the batch. So run() only registers the request and waits on its own queue, and one background loop, started by start(), decides everything:

async def run(self, plan, arrival):
    stream = _Stream(next(self._ids), arrival, plan)
    self._streams[stream.id] = stream
    heapq.heappush(self._arrivals, (arrival, stream.id, stream))
    self._wake.set() # wake the step loop
    try:
        while (item := await stream.queue.get()) is not _DONE:
            index, sim_time, queue_time = item
            lag = max(0.0, self._clock() - (arrival + sim_time))
            yield TokenEvent(index, plan.tokens[index], sim_time, lag, queue_time)
    finally:
        self._policy.abort(stream.id) # free its KV blocks if the client left

async def _steps(self): # simplified
    end = float("-inf") # simulated time the last step ended
    while True:
        # (when idle: wait for a request, then start at its arrival, not now)
        while self._arrivals and self._arrivals[0][0] <= end:
            _, _, s = heapq.heappop(self._arrivals) # only requests that have arrived
            self._policy.add(s.id, s.arrival, s.plan.prompt_tokens, len(s.plan.tokens))

        plan = self._policy.schedule(end) # admit, batch, preempt
        cost = self._cost.step_cost(plan.shape) # roofline: max(compute, memory) + overhead
        end += cost.duration # anchored: lateness never shifts the timeline
        step_lag = await sleep_until(end, self._clock, self._sleep)
        await asyncio.sleep(0) # an overdue loop still yields

        result = self._policy.complete(plan)
        for seq in result.emitted: # one token to every sequence in the batch
            self._streams[seq.id].queue.put_nowait((seq.generated - 1, end - seq.arrival, ...))
        for seq in result.finished:
            self._streams[seq.id].queue.put_nowait(_DONE)
        self._on_step(self._stats(plan, cost, step_lag)) # metrics, via a callback

The output length goes to the policy only so it knows when to stop the sequence, like an end-of-sequence token. Admission never looks at it, because a real server can’t know it.

Both engines share one helper, sleep_until, and that is where the timing model lives.

The timing model: deadlines, not sleeps

The obvious way to fake a token stream is sleep(ttft), then sleep(itl) per token. It has two problems:

Overhead leaks into the metrics. Parsing, validation and matching happen before the first sleep, so observed TTFT becomes overhead plus simulated TTFT, and that overhead varies by prompt.
Errors accumulate. Each sleep wakes slightly late. Over thousands of tokens the lateness adds up and throughput drifts below the configured rate.

Instead, the middleware stamps arrival as early as possible, and the engine sleeps until absolute deadlines:

deadline_i = arrival + ttft + i * itl
await asyncio.sleep(max(0, deadline_i - time.monotonic()))

Overhead is absorbed whenever it is shorter than the simulated time to the first token, and lateness never accumulates, because every token targets its own deadline.

Lag is the simulator’s error bar. When a deadline has already passed, the token goes out at once and the engine records lag = release time − deadline. Lag measures how faithfully the simulator carried out its own timeline, not anything about the simulated GPU. Since only the last token's lag reaches end-to-end time, late tokens catch up rather than stretch the request.

Time is measured in three places , and each layer measures its own:

The gap between server and client is delivery cost; the gap between simulated and server is the simulator running late. Keeping them apart is what lets every result later in this article say which one it is about.

The scheduler: continuous batching on one simulated GPU

The fixed-rate engine treats every request as if it had a GPU to itself. Real servers don’t. Requests share a batch, compete for KV-cache memory, wait when it is full and get evicted when it runs out. The scheduler engine models that, following vLLM's rules.

KV capacity is computed at startup from the hardware profile and the model:

With 80 GiB of HBM, 90% utilisation, a 6.74B-parameter fp16 model (13.48 GB of weights) and 512 KiB of KV per token, that is 7,609 blocks of 16 tokens: about 121,700 tokens, or roughly 29 concurrent requests at 4k context.

Then, every step:

Admit. Waiting requests join in arrival order (preempted ones first) while their whole prompt fits, keeping 1% of blocks free as a watermark, and while fewer than max_num_sequences are running.
Batch. Each running sequence contributes one decode token, or its next prefill chunk. New prompts fill the rest of the max_batch_tokens budget. Prefill is always chunked, so a long prompt spreads over several steps while every other stream keeps decoding.
Grow. A sequence needs a new block every 16 tokens. If none is free, the newest admitted sequence is preempted: its blocks are freed and it goes back to the front of the queue to recompute everything it had.
Duration. The step takes as long as its slowest resource, a roofline: max(compute, memory) + step_overhead, where compute is 2 × parameters × scheduled tokens over peak FLOPS, and memory is the weights plus every sequence's KV over memory bandwidth. Both can be scaled by calibrated efficiencies.
Emit. Sleep until the step’s absolute deadline, then release one token to every sequence whose tokens are computed.

One rule I held to: admission never uses a plan’s real output length. The simulator knows it, but a real server cannot, so it is used only to stop the sequence, like an end-of-sequence token.

How KV pressure reaches the client:

In the example above, one short request decodes at about 4.5 ms per token. Twenty-nine requests at 4k context read about 62 GB of KV per step on top of the weights, so ITL rises to about 25 ms for everyone. Decode is memory-bound; compute takes well under a millisecond.

Code layout. The scheduler has three tiers, with dependencies pointing down only. sequence.py, kv_cache.py and cost_model.py hold resources and cost; policy.py makes the decisions; driver.py is the only code that touches time and asyncio. Because the core is pure, I can run it in virtual time, with no server and no clock, to compute exactly what a load test should produce. That turned out to be the most useful design decision in the project.

Building a benchmark I could trust

My first throughput estimate, 5,500–6,000 output tokens per second, came from an ad-hoc script. It was partly measuring the script. So before load-testing the scheduler I built a load generator, llmsim-bench, that works against any OpenAI-compatible server and checks itself.

Three things mattered most:

Open loop, not just closed loop. A closed loop keeps N requests in flight, sending the next only when one finishes, so a slow server is offered less load and queueing stays hidden. An open loop sends requests on a seeded Poisson schedule whether or not earlier ones have finished, like independent users.
Latency from the scheduled arrival, not from the send. If the client caps concurrency, an arrival may wait for a free slot. Measuring only from the actual send hides that wait, a trap known as coordinated omission. In one run at about 82% utilisation, TTFT p99 was 12 ms from the send and 65 ms from the scheduled arrival, because 66 requests had queued for a slot.
The client checks its own health. A monitor wakes every 10 ms and records how late each wake-up is, plus the client’s CPU. Above 5 ms at p99, the report warns that the client, not the server, inflated the latencies. It caught a real problem in my own tests: with the server in a thread of the test process, both shared the GIL and the client’s loop ran 10 ms late at p99.

A few smaller choices:

It uses aiohttp, not the OpenAI SDK, which builds a typed object for every chunk.
Each synthetic prompt is distinct, and exactly N tokens long, so a real server’s prefix cache cannot flatter TTFT.
After warm-up it resets the server’s /metrics/summary, and after the run it reads it back. Client, server and simulated values then cover the same requests and can sit in one table.

First limits: how far one process can go

Before the scheduler existed, I load-tested the fixed-rate engine to find how much traffic one Python process can stream before it stops keeping time. The simulated TTFT was 7.12 ms and TPOT 4.00 ms; the question was how far the observed values drift from those.

Up to 200 requests per second, about 13,000 output tokens per second, lag p99 stays under 4 ms. Past that, the TTFT tail grows first (p99 49 ms at 250 req/s), and by 350 req/s the median falls behind too. Late tokens then go out back to back, so TPOT reads below the configured 4 ms while end-to-end time stays within 1–4% of schedule.

Two fixes got it there:

uvloop and httptools. uvicorn’s defaults are a pure-Python HTTP parser and the standard asyncio loop. Switching to uvicorn[standard] cut mean lag at 200 streams from 122 ms to 18.6 ms, and mean TTFT from 183 ms to 56 ms.
A regex pre-filter. With a realistic responses.yml, matching every prompt against every pattern collapsed the server at about 60 req/s. Matching held the GIL for about 88% of the time at 60 req/s and stalled the event loop. The fix parses each regex once at load time for the plain strings every match must contain, then skips the regex when the prompt lacks them, with a substring test. Matching a 512-token prompt fell from 10.7 ms to 0.47 ms, and the server held 300 req/s at +2.6 ms median TTFT.

The takeaway: generating a token was never the bottleneck. In-process, the engine releases a token in about 0.7 µs, and tracing plus SSE serialisation bring it to about 9 µs. Delivering it over HTTP costs about 170 µs. A real GPU replica produces a few thousand to low tens of thousands of tokens per second, so one process can stand in for one replica; beyond that, run more processes, as you would run more servers.

Validating the scheduler: predict, then measure

Six of seven experiments passed, most within 0.1–3% of a number written down before the run.

A load test that only describes what happened can’t tell you whether the simulator is right. So each experiment checks one scheduler decision against a prediction, with four rules:

Predict before running. Each prediction comes from the cost model and the scheduler’s rules: a formula when one is enough, otherwise the pure scheduler core run step by step in virtual time.
One variable at a time. Each test model differs from the base model in exactly one field, and a unit test checks that it does.
When a prediction fails, stop. Work out whether the prediction or the simulator is wrong, and never adjust a prediction to fit a result.
An explanation is not proof. An explanation that fits the numbers stays marked as inferred until a control experiment removes the suspected cause.

The base model is the 6.74B one from the scheduler section: 13.48 GB of weights, 512 KiB of KV per token, 3,000 GB/s and 989 TFLOPS. Every response has an exact length, so --max-tokens sets output length precisely.

What each passing result means for someone testing a gateway:

Batching (E1). Sixty-four streams get 38× the throughput of one, for only 1.56× the step time, because decode is memory-bound and the weights are read once per step for the whole batch.
Context (E2). At the same batch size, 16× the prompt length makes every stream’s step 2.8× slower. Long-context users slow everyone down.
Chunked prefill (E3). One long prompt arriving among 16 decoding streams stalls them all for one step. Cutting the token budget from 8,192 to 2,048 lowered that stall from 112 ms to 28 ms, and served the same prompts and decode tokens in the same time. This is the trade-off behind vLLM’s max_num_batched_tokens.
KV admission (E4). With a small cache and 2,000-token prompts, the KV cache, not the batch limit of 64, decides how many run: 9. TTFT becomes almost all queueing, 4.46 s of 4.57 s.
Preemption (E5). Sixty-four requests that outgrow the cache together force about 130 preemptions. The 1% watermark cut them by 9% at no cost in time. The median ITL was 7.9 ms, but a few preempted streams froze for up to 50 s. That is exactly the case a gateway’s timeouts should be tested against.

What the predictions missed

The misses taught me more than the passes. Two stood out.

1. Fixed-length benchmarks phase-lock. In E1, step times matched, but simulated TTFT grew to about 100 ms at 64 streams with nothing blocked. Every request had the same length and the loop was closed, so streams that started together finished together and sent their next requests together. About every 200 steps, one step prefilled most of the batch’s new prompts at once. The largest observed steps were 74–88% of a fully synchronised prefill at every level.

A real engine under the same benchmark would phase-lock too. It suggests that fixed-length, closed-loop benchmarks, a very common setup, overstate TTFT compared with real traffic. For now this is inferred, not proven. If phase-locking is the cause, TTFT p50 should drop to about 10–15 ms.

2. My own request handling leaked into simulated time. E7 swept open-loop load from 10 to 100 req/s. Batch size and step time matched at every rate, but two results failed:

TTFT was 3.5–5 ms higher than predicted at every rate. The arrival is stamped before the body is parsed and the generator runs. An idle engine absorbs that time; a busy one can’t add a request to a step that is already running. So the processing time, rounded up to the next step boundary, shows up as queue time.
Capacity was 9% lower: 74 req/s against 81.7. New prompts reached the engine about three at a time instead of one per step. One 512-token prompt riding with 64 decodes is free: its 7.9 ms of compute hides under the step’s 10.6 ms of memory time. Three together make the step compute-bound at 21.8 ms. Feeding the virtual run the same arrivals in 20–30 ms bunches reproduced the real run on every measure, capacity 70–77 req/s, which confirms the bunching is enough to explain the gap.

Neither is a scheduling bug. Both are the cost of the HTTP process in front of the scheduler, which is exactly the part of the system I had not yet optimised.

Where the time goes

With the scheduler, lag grows with batch size, not with throughput: token lag p99 is about 1.7 ms plus 0.15 ms per sequence in the batch.

At 64 streams that is only 3,100 output tokens per second, a quarter of what the fixed-rate engine sustains within the same 5 ms bar. The cause is the shape of the work. The fixed-rate engine spreads each request’s tokens over its own timeline. A scheduler step releases one token to every sequence at the same instant, as a real engine does, and one event loop then delivers them one after another. Tokens late in that queue show it as lag.

Simulated values are unaffected. The timeline is anchored, so step times and simulated TTFT and E2E are exact at every level; observed ITL p50 matched the simulated step time (16.72 vs 16.82 ms at 64 streams). Lag only adds jitter to what clients observe.

To see where each token’s CPU goes, I profiled 64 streams with py-spy:

One lesson about the instrument itself. py-spy’s default mode pauses the process for every sample, and under this load it inflated CPU per token 1.7× and doubled the step loop’s share. I trusted the profile only after it matched two independent measurements: process CPU from /proc (89–91 µs per token) and timers around the step loop inside the server (13.4 µs).

The breakdown also sets realistic expectations for the fix. Prebuilt SSE templates (about 22 → 3 µs) and lighter per-token metrics (about 15 → 7 µs) bring the total to about 60 µs.

The socket write, uvloop and the task switch, about 25–30 µs together, remain. So lag p99 should roughly halve, and observed timing should stay within 5 ms up to about 40–50 streams instead of 25.

Known limits and what’s next

On the same workload (512-token prompts, 64 output tokens), one process keeps observed timing within 5 ms of the simulated timeline up to:

about 200 req/s with the fixed-rate engine, about 13,000 output tokens per second;
about 25 req/s with the scheduler, a batch of about 10 and 1,675 output tokens per second.

Beyond that, simulated step times, batches and preemptions stay right, but observed per-token times carry the process’s own overhead, and the bench report flags it. Lag is always measured, so a result never silently drifts outside the envelope.

Next, I would study:

Running an experiment to confirm or rule out phase-locking.
Prebuilt SSE templates and lighter per-token metrics , measured the same way as the breakdown above, with a test that the bytes sent stay identical.
gc.freeze() after startup. One full garbage collection took 13.7 ms, the largest lag outlier.
Further out: prefix caching in the scheduler, tensor parallelism, several processes as several replicas, and a virtual-time mode for capacity planning without real sleeps.

Try it

docker/bench.sh

That builds the images, starts a simulator and runs a rate sweep against it, writing JSON results to results/. Point any OpenAI client at http://localhost:8080/v1 and it works unchanged.

The takeaway: the scheduler’s rules were the easy part to get right; the predictions matched within a few percent. What the predictions missed was the HTTP process in front of it, the same layer real serving stacks spend so much effort on.

A fun exercise in writing the number down before putting load on the system, and then finding out why it was wrong.

Source Code & Full Results:

GitHub - umangaggarwal2/llm-simulator: A deterministic fake LLM that streams tokens through OpenAI compatible API, provides custom GPU configurations and simulates metrics...

Recipe of Near Real-Time Retrieval

Umang Aggarwal — Sun, 13 Sep 2026 07:53:06 +0000

In this article, we will explore a search & retrieval design for natural language queries, expecting data in real-time consistency with updates. We will build a write-path, segregate the read-path and a streaming pipeline in cloud. We will also look into the scalability and performance of the architecture.

The complete Git repository for the benchmarking is provided at the end of this article.

To make a retrieval mechanism over data being updated in real-time, we need the ingredients:

Updates on a database
Real-time streaming, indexing and inference
Natural queries expecting updated output

Everytime we make an update in our database, we will asynchronously replicate or stream the write to our search engine.

Building the streaming engine

Kafka

Kafka is a distributed append-only log that uses a publish-subscribe model to let producers write data streams and consumers can read them independently. It separates the data streams into topics, distributed partitions and consumer groups, and retains them on disk based on topic specific retention policies, which enables multiple consumers to read data simultaneously with each having their independent lag. The tolerance of lag makes Kafka the correct message bus for streaming our updates to the retrieval engine.

Debezium

Any relational database first writes the updates to a log before committing the write to the storage — Write Ahead Log. Reading these logs and copying these writes somewhere else is called replication. Debezium is a set of source connectors designed for the Kafka Connect framework. It reads the logical WAL of the database, which is designed for its own replication and PITR (point-in-time-recovery), and translates it to a Kafka readable message. This process is called Change Data Capture (CDC).

Building a search engine

Keyword Search

Given a query with keywords “A”, “B” & “C”, we can return the data with the highest frequencies of these keywords, thus making them more relevant. This simple approach is called a bag of words and the most commonly used function to calculate the score is Okapi BM25. Querying requires an Inverted File Index , in which keywords are mapped with a list of document IDs along with its term frequency. This index is implemented at scale by Lucene — the engine behind ElasticSearch and OpenSearch.

Semantic Search

Keyword search works well when the query and the document share the same words. However, consider a query such as

“How do I reserve a seat on a flight?”

A document containing:

“Passengers can temporarily hold an available seat before completing payment.”

is clearly relevant to the query, even though many of the words are different.

Keyword search has no inherent understanding of this relationship. We therefore need a representation of text that captures its meaning, rather than just the words it contains.

This is where embeddings come in.

An embedding model converts a piece of text into a numerical vector reprsesentation:

“How do I reserve a seat on a flight?”

↓ embedding model

[0.12, -0.34, 0.87, …, 0.21]

The individual numbers are not particularly meaningful to us. What matters is that text with similar meaning tends to produce vectors that are close together in the embedding space.

“How do I reserve a seat?”

↓

[0.12, -0.34, …]

“Can I temporarily hold a flight seat?”

↓

[0.10, -0.31, …]

These vectors can then be compared using a similarity metric such as cosine similarity.

This gives us a fundamentally different search mechanism:

Keyword search:

query → terms → inverted index → matching documents

Semantic search:

query → embedding → vector similarity → similar documents

Vector Databases

To power the query for vectors, we need specialised databases that operate on Aproximate Nearest Neighbours (ANN).

However, at a scale of millions of vectors, the neighbour calculations may require millions of similarity calculations for every query. Just like keyword search required Inverted File Index, vector databases use specialized indexing structures. A popular approach is Hierarchical Navigable Small World (HNSW).

HNSW organises vectors into multiple layers of a graph. Higher layers provide long-range connections that allow the search to quickly move towards the relevant region of the vector space, while lower layers provide increasingly fine-grained connections. This allows us to find approximate nearest neighbours without comparing the query against every vector.

For our use case, we can use Qdrant as the vector database. It stores the generated embeddings and provides efficient approximate nearest-neighbour search using HNSW index.

Embedding generation then becomes a separate concern, we can use an embedding model hosted by a cloud provider or run a local inference service.

Combining Keyword & Semantic Search

In practice, we need to look for documents that have both similar meaning as well as exact words. Therefore, we will perform both keyword and vector search in parallel and combine the results by using Reciprocal Rank Fusion (RRF).

RRF is a simple algorithm that combines multiple ranked lists into a single unified list. It does not require the scores produced by the different retrieval systems to compare the documents. Instead, it uses the rank of a document in each result list to calculate a combined score.

RRF(d) = Σ 1 / (k + rank(d))

Hence, a document that appears on top of multiple lists is more likely to rank higher in the final set. We can have the detailed implementation later.

For now, our final ETL architecture loads data both into two separate data stores.

Infrastructure

We will split the architecture between managed cloud services, preferrably over multiple regions for network latencies, and lightweight local infrastructure. This will keep the setup simple while allowing us to demonstrate a realistic CDC-driven search pipeline.

PostgreSQL (AWS RDS) acts as the source of truth for a users dataset.

Kafka + Debezium run locally in Docker containers. Debezium reads PostgreSQL’s logical WAL and publishes database changes to Kafka.

Amazon OpenSearch is used for traditional keyword search and BM25 indexing.

Qdrant Cloud stores the document embeddings and provides vector similarity search.

Kafka consumers independently process the CDC stream:

3 consumer processes will consume 3 kafka paritions
we will keep parallel indexing methods for Opensearch and Qdrant indexing.
we could keep separate consumers for each indexing workflow, however, it will be easier to collect benchmark reports from a single script.

Setting up PostgreSQL for CDC

We will use Amazon RDS for PostgreSQL and configure it to expose logical changes through the PostgreSQL WAL.

Create a PostgreSQL RDS instance

Provision a small PostgreSQL instance on AWS.
Configure network access so the local CDC infrastructure can connect to it.

Create the application database and table

The users table contains the fields that will eventually be indexed by both OpenSearch and Qdrant.

CREATE TABLE users (
    id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    name TEXT,
    email TEXT,
    company TEXT,
    job_title TEXT,
    location TEXT,
    skills TEXT,
    bio TEXT,
    experience INT,
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);

Enable logical replication

PostgreSQL needs to retain the logical changes in its WAL so that Debezium can consume them.

On RDS, this involved enabling:

rds.logical_replication = 1

After applying the parameter-group change and restarting the instance, PostgreSQL should report:

wal_level = logical

Create a dedicated replication user

We need to create a replication only user for our CDC connector:

CREATE USER debezium
WITH PASSWORD '...';

GRANT rds_replication TO debezium;

GRANT CONNECT ON DATABASE appdb TO debezium;

GRANT USAGE ON SCHEMA public TO debezium;

GRANT SELECT ON TABLE public.users TO debezium;

Create a publication for the table

PostgreSQL logical replication uses a publication to define which tables and operations are exposed through logical replication.

CREATE PUBLICATION users_cdc
FOR TABLE public.users;

This publication exposes changes to the users table, which Debezium subsequently reads from the PostgreSQL WAL.

Verify the replication configuration

We can now verify the PostgreSQL configuration for our CDC pipeline.

SHOW wal_level;

-- should return `logical`

SELECT *
FROM pg_publication_tables;

-- should contain the expected table `users`

Pre-loading Opensearch and Qdrant

Before setting up our CDC pipeline, we will first populate the AWS Opensearch and Qdrant cloud instances.

For Opensearch, we can use a Logstash pipeline as a convenient way to perform the initial bulk load directly from PostgreSQL.

input {
  jdbc {
    jdbc_connection_string => "jdbc:postgresql://<RDS_ENDPOINT>:5432/appdb"
    jdbc_user => "indexer"
    jdbc_password => "<PASSWORD>"
    jdbc_driver_library => "/path/postgresql.jar"
    jdbc_driver_class => "org.postgresql.Driver"

    statement => "
      SELECT
        id, name, email, company, job_title,
        location, skills, bio, experience,
        created_at, updated_at
      FROM users
    "
  }
}

output {
  opensearch {
    hosts => ["<OPENSEARCH_ENDPOINT>"]
    user => "<USERNAME>"
    password => "<PASSWORD>"
    index => "users"

    # Keep the Postgres primary key as the OpenSearch document ID
    document_id => "%{id}"

    ecs_compatibility => disabled
    ssl_certificate_verification => true
  }

  stdout {
    codec => rubydebug
  }
}

We can then query the Opensearch instance:

GET users/_search
{
  "size": 1,
  "_source": ["id"],
  "query": {
    "multi_match": {
      "query": "backend engineer kafka",
      "fields": [
        "job_title",
        "skills",
        "bio",
        "company",
        "location"
      ]
    }
  }
}

and expect a response as:

{
  "took": 378,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": {
      "value": 8621,
      "relation": "eq"
    },
    "max_score": 2.1194136,
    "hits": [
      {
        "_index": "users",
        "_id": "2970",
        "_score": 2.1194136,
        "_source": {
          "id": 2970
        }
      }
    ]
  }
}

Similarly, we can bulk-load the Qdrant instance using a Python script:

from qdrant_client import QdrantClient, models

client = QdrantClient(
    url="<QDRANT_URL>",
    api_key="<QDRANT_API_KEY>",
    cloud_inference=True,
)

collection = "users_semantic"

points = []

for user in users:
    text = f"""
    Name: {user["name"]}
    Company: {user["company"]}
    Job title: {user["job_title"]}
    Location: {user["location"]}
    Skills: {user["skills"]}
    Bio: {user["bio"]}
    """

    points.append(
        models.PointStruct(
            id=user["id"],
            vector=models.Document(
                text=text,
                model="sentence-transformers/all-MiniLM-L6-v2",
            ),
            payload=user,
        )
    )

client.upsert(
    collection_name=collection,
    points=points,
)

Each Qdrant point uses the PostgreSQL id as its point ID, keeping the identity consistent across the source database, OpenSearch and Qdrant.

Local CDC infrastructure

For the streaming layer, we will use docker compose to run Kafka and Debezium locally. This will make it possible to experiment with the CDC pipeline without having to operate a complete Kafka cluster at cloud.

services:

  kafka:
    image: apache/kafka:4.0.0
    container_name: kafka
    hostname: kafka

    ports:
      - "9092:9092"
      - "9094:9094"

    environment:
      # KRaft - single node
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller

      # PLAINTEXT is for other containers on the "cdc" network (e.g.
      # debezium), which resolve "kafka" via Docker's internal DNS.
      # EXTERNAL is for clients running on the host machine (e.g.
      # consumer/index_users.py via the venv), which can't resolve
      # "kafka" and need "localhost" instead.
      KAFKA_LISTENERS: PLAINTEXT://:9092,EXTERNAL://:9094,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,EXTERNAL://localhost:9094

      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093

      # Single-node development settings
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0

      KAFKA_NUM_PARTITIONS: 3

    volumes:
      - kafka-data:/var/lib/kafka/data

    networks:
      - cdc

  debezium:
    image: quay.io/debezium/connect:3.3
    container_name: debezium
    hostname: debezium

    depends_on:
      - kafka

    ports:
      - "8083:8083"

    environment:
      BOOTSTRAP_SERVERS: kafka:9092

      # Kafka Connect worker
      GROUP_ID: 1

      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_status

      CONFIG_STORAGE_REPLICATION_FACTOR: 1
      OFFSET_STORAGE_REPLICATION_FACTOR: 1
      STATUS_STORAGE_REPLICATION_FACTOR: 1

      # JSON for easy inspection while learning
      KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter

      KEY_CONVERTER_SCHEMAS_ENABLE: "false"
      VALUE_CONVERTER_SCHEMAS_ENABLE: "false"

      # REST API
      REST_PORT: 8083
      REST_ADVERTISED_HOST_NAME: debezium

    networks:
      - cdc

volumes:
  kafka-data:

networks:
  cdc:
    driver: bridge

We can get our Kafka+Debezium cluster up by executing

docker compose up -d

Once the cluster is up and running, we will do a health check and then register our connector:

# Verify the container processes
docker compose ps

# kafka running
# debezium running

# Check that Kafka is responding
docker exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka:9092 \
  --list

# connect_configs
# connect_offsets
# connect_status
# Later once debezium is registered we will also see - 
# usersdb.public.users
# schemahistory.users

# Verify the Kafka Connect internal topics
docker exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka:9092 \
  --describe \
  --topic connect_configs

# PartitionCount: 1
# Configs: cleanup.policy=compact

docker exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka:9092 \
  --describe \
  --topic connect_offsets

# PartitionCount: 
# Configs: cleanup.policy=compact

docker exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka:9092 \
  --describe \
  --topic connect_status

# PartitionCount: 
# Configs: cleanup.policy=compact

Now, we can validate our Debezium instance:

curl http://localhost:8083/

curl http://localhost:8083/connectors

# should return []

curl http://localhost:8083/connector-plugins

# {
# "class": "io.debezium.connector.postgresql.PostgresConnector",
# "type": "source",
# "version": "..."
# }

It is better to check for the debezium logs as well —

docker logs debezium --tail 100
docker logs -f debezium

# We should not see - 
# ConfigException
# Kafka Connect stopping
# Uncaught exception in herder work thread

Registering the PostgreSQL connector requires a connector configuration:

{
  "name": "users-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",

    "database.hostname": "",
    "database.port": "5432",
    "database.user": "",
    "database.password": "",
    "database.dbname": "",
    "database.sslmode": "require",

    "topic.prefix": "usersdb",

    "plugin.name": "pgoutput",

    "slot.name": "users_cdc_slot",
    "publication.name": "users_cdc",
    "publication.autocreate.mode": "disabled",

    "table.include.list": "public.users",

    "snapshot.mode": "no_data",

    "tombstones.on.delete": "false",

    "schema.history.internal.kafka.topic": "schemahistory.users",
    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092"
  }
}

Then we can call the registration API:

curl -X POST \
  -H "Content-Type: application/json" \
  --data @users-cdc.json \
  http://localhost:8083/connectors

# check the connector 
curl http://localhost:8083/connectors/users-cdc

curl http://localhost:8083/connectors/users-cdc/status

# {
# "name": "users-cdc",
# "connector": {
# "state": "RUNNING"
# },
# "tasks": [
# {
# "id": 0,
# "state": "RUNNING"
# }
# ]
# }

Indexing Consumer

The consumers read CDC events from Kafka and translate them into operations on the respective search systems.

For each database change:

INSERT → create/upsert the document
UPDATE → update/re-embed the document
DELETE → remove the document from the indexes

This also makes the pipeline change-driven : instead of periodically scanning the entire database, only the records that changed need to be propagated.

if operation in ("c", "u", "r"):
    user = payload.get("after")

    opensearch_future = executor.submit(
        index_opensearch, user
    )

    qdrant_future = executor.submit(
        index_qdrant, user
    )

    wait([opensearch_future, qdrant_future])

For Opensearch we will upload the complete document:

document_id = str(user["id"])

opensearch.bulk(
    body=[
        {
            "index": {
                "_index": "users",
                "_id": document_id,
            }
        },
        user,
    ]
)

For Qdrant we will build the search text and then publish:

def build_search_text(user):
    fields = [
        user.get("name"),
        user.get("job_title"),
        user.get("company"),
        user.get("location"),
        user.get("skills"),
        user.get("bio"),
    ]

    return " ".join(
        str(value)
        for value in fields
        if value
    )

models.PointStruct(
    id=int(user["id"]),
    vector=models.Document(
        text=build_search_text(user),
        model="sentence-transformers/all-MiniLM-L6-v2",
    ),
    payload=user,
)

The complete consumer script is shared at —

real-time-retrieval/consumer/index_users.py at master · umangaggarwal2/real-time-retrieval

Querying: Hybrid Search

We are finally at the last piece of our architecture — retrieval. We will use both keyword and semantic search, i.e. retrieve from Opensearch and Qdrant and combine the results using RRF.

The steps for writing a retrieval script are —

BM25: top 50
semantic: top 50
BM25 weight: 0.6
semantic weight: 0.4
k = 60
final results: top 10

Where the RRF score is defined as:

score(d) =
    0.6 * 1 / (60 + bm25_rank(d)) +
    0.4 * 1 / (60 + semantic_rank(d))

RRF lets the two retrieval systems contribute based on rank rather than trying to directly compare BM25 scores with cosine similarity scores, which live on different scales.

Find the complete querying script at —

real-time-retrieval/search/query.py at master · umangaggarwal2/real-time-retrieval

Benchmarking the performance

We may have the system ready and working.

But this architecture becomes much more interesting when we understand how it behaves under load — and, more importantly, how much “real-time” we actually get.

We will benchmark the write and read paths with a target load of 100 QPS for 2–5 minutes.

Peformance — defines how fast a system works for a single task or at a pre-defined benchmark scale.

Hence, we will measure the latency, while keeping our benchmark 100 QPS, but for a small dataset of 10K records.

Rather than measuring only end-to-end latency, we instrument each stage of the pipeline so that we can distinguish database latency, CDC lag, queueing, network overhead, and downstream indexing/inference time.

Write → database
Database WAL →CDC
CDC → Kafka
Kafka → Consumer
Consumer → OpenSearch Indexing: including network overhead and indexing latency
Consumer → Qdrant Indexing: including network overhead, embedding generation and indexing latency

# Write benchmarking
def run_writer(csv_rows, rate, duration, concurrency):
    def _one_update():
        # Pick an existing target user.
        target_row = random.choice(csv_rows)
        target_id = int(target_row["id"])

        # Pick another row as the source of new values.
        source_row = random.choice(csv_rows)

        return perform_update(
            _pg_worker.cursor,
            target_id,
            source_row,
        )

    def _on_success(result):
        user_id, updated_at, db_latency_ms = result

        # Convert PostgreSQL timestamp into an epoch timestamp.
        if updated_at.tzinfo is None:
            updated_at = updated_at.replace(tzinfo=timezone.utc)

        update_time = updated_at.timestamp()

        with state.lock:

            state.total_updates += 1

            state.db_latencies_ms.append(db_latency_ms)

            state.pending[user_id] = {
                "updated_at": update_time,
                "created_at": time.perf_counter(),
            }

            state.all_target_ids.append(user_id)

# Read benchmarking
def run_benchmark(
    queries,
    rate,
    duration,
    warmup,
    concurrency,
):
    def _one_query():
        query_text = random.choice(queries)
        return query_text, query(query_text)

    def _on_success(result):
        query_text, result_dict = result

        latency = result_dict["latency"]
        overlap = result_dict["overlap"]

        with state_lock:
            bm25_latencies.append(latency["bm25_ms"])
            semantic_latencies.append(latency["semantic_ms"])
            rrf_latencies.append(latency["rrf_ms"])
            total_latencies.append(latency["total_ms"])
            overlap_percentages.append(overlap["percentage"])

Write Performance: P99

PostgreSQL write: 762.62 ms
PostgreSQL -> OpenSearch: 65431.88 ms
PostgreSQL -> Qdrant: 65431.88 ms

Write Latency Breakdown:

Postgres commit -> Debezium capture: 10906.02 ms

Debezium -> consumer receive (Kafka): 302706.23 ms

OpenSearch: network overhead: 543.66 ms

OpenSearch: server-side indexing: 14.00 ms

Qdrant: network overhead: 261.87 ms

Qdrant: server-side embed + HNSW (time): 192.16 ms

Read Performance: P99

OpenSearch / BM25 latency (total round-trip, network + server): 7634.46 ms

Qdrant / semantic latency (total round-trip, network + server): 2107.44 ms

RRF computation latency: 0.29 ms

End-to-end hybrid latency: 8517.81 ms

BM25 / semantic top-50 overlap: _ **_40%**

Bottlenecks

The benchmark gives us a few interesting observations about where the system starts to struggle under load.

1. Consumer throughput is the bottleneck in the write path

We are generating around 100 database updates per second , while the indexing consumer can process only around 1–3 updates per second per instance.

Hence, the database is able to accept the writes, but the downstream consumer is not able to process the changes at the same rate.

This creates a backlog in Kafka.

2. OpenSearch is the bottleneck in the read path

The read benchmark gives us a different bottleneck.

We are querying OpenSearch for BM25 results and Qdrant for semantic results and then combining both using RRF.

The p99 latency for OpenSearch was: 7634.46 ms

The RRF computation is negligible compared to the time spent retrieving the results. Qdrant also remains significantly healthier than OpenSearch under this workload.

The OpenSearch instance starts struggling when we increase concurrent BM25 queries. This results in request timeouts and causes the achieved query throughput to fall significantly below the requested 100 QPS.

Hence, increasing the application concurrency alone does not solve the problem. The bottleneck is the query capacity of the OpenSearch deployment.

For a production system handling this level of traffic, we would need to increase the capacity of the search infrastructure rather than simply increasing the number of application threads.

Repository:

https://github.com/umangaggarwal2/real-time-retrieval/