shanhui notes
job workers queue in postgres and learning Rust
this is part of the multi-part series on building a jobs search engine in Rust. in the first part i wrote about the mechanics of how vector similarity search with embeddings work. here i will discuss building background task queues using rust async primitives and some learnings.
the embeddings indexer is designed to be independent of the job post parsing pipeline. job post parsing might be done in high volume such as by an web crawler or one off process by a company submitting a job post via an email worker. when a new job listing gets created, it has a pending embedding_status. the job embeddings indexer scans this table for pending states and also embedded states that have an updated_at after the embedded_at datetime.
SELECT ... FROM job_listings
WHERE embedding_status = 'pending'
OR (embedding_status = 'embedded' AND updated_at > embedded_at) -- ← re-embed trigger
ORDER BY created_at
LIMIT 1
FOR UPDATE SKIP LOCKED
we implement the EmbedIndexWorker as tokio task loop in the instance as the http api service.
// main.rs
let worker_handle = {
let worker_service: JobListingService = JobListingService::new(pool, config.deepinfra_api_key.clone());
let worker = EmbedIndexerWorker::new(worker_service, config.indexer);
let shutdown: CancellationToken = shutdown.clone();
tokio::spawn(async move {
worker.run(shutdown).await;
})
};
inside the loop we use tokio::select!, select! polls multiple futures concurrently and takes the first on that completes. the other paths are cancelled. this construct is used to enable graceful shutdowns.
select! {
_ = sleep!(self.config.poll_interval) = {}
_ = shutdown.cancelled() => {
tracing::info!("embed indexer shutdown signal received");
break;
}
}
we use a match inside an infinite loop for pattern matching.
loop {
match self.jobls_service.embed_next_pending().await {
Ok(true) => {
select! {
...
}
}
Ok(false) => {...}
Err(JobEmbedError::Embedding(ref e)) if e.is_transient() => {...}
Err(other) => {...}
}
}
it is designed so that the task happens outside the select! {} because Rust async is cooperative, not preemptive, you can not kill an awaiting future mid flight from the outside. if the work is done inside the select block, then a unexpected shutdown would cancel the work, possible during an HTTP request for DB insert which would have unintended effects.
by putting the work before the select!, canceling mid-work only matters during the sleep, once the work starts, it finishes, and only the next idle await is affected by the shutdown signal which enables graceful shutdowns and restarts.
instead of draining the queue as fast as possible creating sharp spikes for downstream services, this indexer runs every n=30 seconds default to spread load across the day. this of course sets a maximum amount of embeddings per day of 2880 listings that can be embedded per day which is more than enough for my intended load. this env variable can be lowered if there is more load.
atomic updates
we persist the the embeddings atomically, and set the job listing embedded_status to embeded and embedded_at to NOW(), inserting the vector and flipping the status in one transaction is what prevents the status to be stuck in pending with vectors already created.
error handling for embeddings indexer
there are 2 kinds of errors which we classify into transient and non transient errors. transient errors include API temporary hiccups, auth, or rate limits. the workers rolls back the transactions, returns Err and retries using an exponential backoff 30s * 2^attempt, capped at 120s.
pub fn is_transient(&self) -> bool {
match self {
HttpError(_) | IoError(_) | EmptyResponse => true,
ApiError { status, .. } => matches!(*status, 401 | 403 | 429) || *status >= 500,
JsonError(_) => false,
}
}
some notes of learning and using Rust
i had to learn Rust to build this search engine project. if it was pre-AI, it would be one hell of a task and would probably take a few weeks if not months before i can even get started into the crux of the system architecture. but with tools like Cursor and Pi, you can just jump right into it and learn as you build stuff which is a lot more efficient than spending hours on cookie cutter tutorials and hello world examples.
one of the things i immediately find enjoyable about Rust is the control flow that i come to know with terse functional style Typescript or Python code.
fn hits_from_rows(rows: Vec<JobListingsSearchRow>, size: usize) -> Vec<JobSearchHitResponse> {
rows.into_iter()
.take(size)
.map(|row| JobSearchHitResponse {
listings: JobListingResponse::from(row.listing),
score: score_from_cosine_distance(row.distance),
})
.collect()
}
// Compact top-N summary for rerank eval logs (vector order)
fn format_vector_top_for_log(rows: &[JobListingSearchRow], n: usize) -> String {
rows.iter()
.take(n)
.enumerate()
.map(|(i, row)| {
format!(
"{}. {} | {} | id={} | distance={:.4}",
i + 1,
row.listing.company_name,
row.listing.job_title,
row.listing.id,
row.distance
)
})
.collect::<Vec<_>>()
.join(" ; ")
}
in rust, you can with ease iterate over String chars using iterators. here we iterate over the Char iterator to create slug.
fn slugify_parts(s: &str) -> String {
let mut out = String::with_capacity(s.len());
let mut last_with_hyphen = false;
for c in s.chars() {
// ...
}
out.trim_matches('-').to_string()
}
you can use ? to chain and handle early returns for any value of type Option<T> or Result<T, E>. if the result is None or Err(e), we can immediately return. if there is Some value, we continue.
fn single_header_ip(headers: headers: &HeaderMap, name: &str) -> Option<IpAddr> {
headers.get(name)?.to_str().ok()?.trim().parse().ok()?
}
with the ? helper, you would write multiple layers of matches or towers of and_then which is a lot more verbose.
fn single_header_ip(headers: &HeaderMap, name: &str) -> Option<IpAddr> {
match headers.get(name) {
None => None,
Some(v) => match v.to_str().ok() {
None => None,
Some(s) => s.trim().parse().ok(),
},
}
}
rust has great pattern matching support and succinct conventions to map data structs to another data struct using the From trait.
impl From<JobListingModelShape> from JobListingResponse {
fn from(m: JobListingModelShape) -> Self {
let salary_range = match (m.salary_min_ks, m.salary_max_ks) {
(Some(min: i32), Some(max: i32)) => Some(format!("${}K - ${}K {}", min, max, m.currency_code)),
(Some(min), None) => Some(format!("From ${}K {}", min, m.currency_code)),
(None, Some(max)) => Some(format!("Up to ${}K {}", max, m.currency_code)),
_ => None,
};
let company_name: &str = &m.company_name;
let job_title: &str = &m.job_title;
let city_location: &str = &m.city_location;
let detail_page_slug: String = make_detail_page_slug(company_name, job_title, city_location, &m.id);
Self {
salary_range,
... // rest of JobListingResponse fields
}
}
}
in Rust, you do not need the return keyword for final values in a fn similar to Ruby or Elixir.
the other thing i like about Rust, is that unit tests are in the same file as the implementation code which makes tests seem more like a first class concern than other languages like Python or Typescript.
i had to struggle with the rust analyzer in Cursor for a bit because it kept highlighting errors that the Rust compiler did not consider an error. i fixed it by pointing the IDE rust analyzer to use the cargo checker directly in Cursor settings. you can just ask your coding harness to point the rust analyzer at the installed Rust Cargo binary and it should fix the issue.
overall, find Rust to be a lot more approachable with current AI tools and i see many more projects migrating to it for the memory safety guarantees, strong type checking, and helpful compiler. this unit tests are also where the implementation code is and treated as a first class principle. the compile times are a bit slower than my experiences with Go and Typescript, but for all that the compiler does, i think it is a fair tradeoff for the benefits of a strong compiled language. i used Grok 4.5 and Deepseek V4 Flash with Rust and find both models to be well versed in Rust programming conventions and concepts.