feat: cyclotron

[oliverb123]

Problem

We want our delivery system (aka "the CDP" or "destinations" or probably other names too, I find it hard to keep up) to deliver (insert very large number here) of events per second. This is the start of a PG-based, sharded job queue system that's intended to let us do that (while managing QoS on a per-user or per-endpoint basis, and doing other fancy tricks kafka makes difficult). The underlying implementation is written in rust, but it's designed to be easy to expose bindings to other languages, so projects or teams favouring TS/JS (e.g. the hog folks) or python (I haven't been in the company long enough to call out anyone specific) can still interact with our "delivery engine" (queue work, manage work for a given team or function or endpoint or queue or whatever, or even ship a worker to consume jobs from the queue, if needed).

Right now only the PG part is in place here, the sharding is mostly signs in the ground labelled "draw the rest of the F'ing owl", and the management commands I first ship with will be executed directly on the DB (or DBs) by the issuing manager instance, even though the intention is for them to be pushed onto a kafka queue and for each shard to individually manage it's control tables. Also I need to do a bunch of query optimisation (although one of the nice things here is that, because the DB is sharded, we can afford to be a bit inefficient with our DB ops, since if we need more throughput we just spin up more shards), and so on and so on and so on. It's v0, it's going to be a little rough around the edges.

You might notice this looks a bit like rustyhook. You'd be correct, we stole a fair amount of both inspiration and literal SQL from it while sketching this out. We didn't try to simply extend rustyhook because the queue implementation there made a series of design choices that make it unsuitable for 1) being embedded in other languages 2) having more than 1 type of worker operate atop it and 3) supporting the kind of work-management features we expect the delivery system to need. These were good decisions - rusty hook was designed to only use it's own queue internally for retries, needed to ship, and was /definitely/ never meant to be embedded in other languages - but our delivery solution is maturing and growing in complexity, and it's needs have outgrown rustyhook. Onwards and upwards.

[bretthoerner]

Looks like a good starting point, nice work.

I have to admit after seeing the code it's not super obvious what Rust is doing for us. 😓

But I guess being able to write workers like fetch in it will be nice.

[bretthoerner]

total nit, but this seems more like a PR comment than a README

[bretthoerner]

You'd only have to bump it for producers/consumers of that queue anyway, right? Seems fine.

[bretthoerner]

Just to note, as discussed on the call, if we think many jobs will run long enough to require a heartbeat (beyond the initial dequeue) I think we'd save a lot by heartbeating a single session (unique per worker process).

[bretthoerner]

Or the entire row could be written outside of the DB, even just to logs in the short term.

Anyway, this is a good idea. Poison pills really hurt us in the txn-based Rusty-Hook, and this will be a nice safety net.

[bretthoerner]

Yeah, it could be nice for the janitor to check some of our invariants and report anything suspicious. Doesn't have to be now though.

[bretthoerner]

Interesting, I'll have to read more code but it's surprising to me that a worker would even have lots of little updates to do for a job, rather than a single simple update at the end.

[oliverb123]

Most wont, but e.g. in the fetch impl, I have a pattern of queuing up the set of updates that would put a job into the deal letter queue, then doing some serde or other "this should never fail, if it does that's a coding error in the fetch worker" work, and if any of that causes an error I just flush the update (sending the job to the DLQ), otherwise I then queue up the updates to send it back to an available state or whatever, and flush

It looks like (ripped from in-progress code, but you maybe get the idea)

// Complete the job, either because we got a good response, or because the jobs retries
// have been exceeded.
pub async fn complete_job(
    worker: &Worker,
    job: &Job,
    return_worker: WaitingOn,
    return_queue: Option<String>,
    on_finish: OnFinish,
    result: FetchResult,
) -> Result<(), WorkerError> {
    // If we fail any serde, we just want to flush to the DLQ and bail
    worker.set_state(job.id, JobState::Available)?;
    worker.set_queue(job.id, DEAD_LETTER_QUEUE)?;

    let is_completed = result.is_completed();

    let result = match serde_json::to_string(&result) {
        Ok(r) => r,
        Err(e) => {
            // Leave behind a hint for debugging
            worker.set_metadata(job.id, Some(format!("Failed to serialise result: {}", e)))?;
            worker.flush_job(job.id).await?;
            return Err(WorkerError::SerdeError(e));
        }
    };

    worker.set_queue(
        job.id,
        &return_queue.unwrap_or_else(|| job.queue_name.clone()),
    )?;
    worker.set_waiting_on(job.id, return_worker)?;

    match (is_completed, on_finish) {
        (true, _) | (false, OnFinish::Return) => {
            worker.set_state(job.id, JobState::Available)?;
        }
        (false, OnFinish::Complete) => {
            worker.set_state(job.id, JobState::Failed)?;
        }
    }

    worker.set_parameters(job.id, Some(result))?;
    worker.set_metadata(job.id, None)?; // We're finished with the job, so clear our internal state
    worker.flush_job(job.id).await?;

    Ok(())
}

[bretthoerner]

Just my opinion, but following what state everything in here seems like more of a burden than just having a happy path that sets things at the end, otherwise throw an error, and have a wrapper function that catches errors and sets the state properly in that case.

Being N lines into a function and having to remember "OK, at the start we set it to DLQ at fn start, so if X happens then Y will happen" only gets worse over time in my experience.

[bretthoerner]

Kind of like above, I'm still confused why our whole update API isn't just a "I'm done with the job, here is its new state" type call.

[bretthoerner]

Suggested change

	// a since chunk of bytes). These are convenience functions for converting between
	// a single chunk of bytes). These are convenience functions for converting between

(This kind of trails off, too.)

[github-actions]

Size Change: 0 B

Total Size: 1.08 MB

ℹ️ View Unchanged

Filename	Size
frontend/dist/toolbar.js	1.08 MB

_{compressed-size-action}