DEV Community: Falon Darville

Simple IT: Set Up and Ongoing Maintenance of Zero Assumptions

Falon Darville — Sat, 29 Apr 2023 20:35:47 +0000

Zero Assumptions is easy to set up and easy to maintain. A very small team or even a single person is enough to get Zero Assumptions up and running. The benefit with our solution is that the cost is very low - and we've considered the in-house costs beyond the payment for our product. It's low cost because it won't take a whole team to maintain.

Think of Zero Assumptions as barely perceptibly, and immensely impactful. You likely already have someone on staff who can set up and maintain it. This minimally invasive monitoring tool can be launched in a day given that servers are ready to funnel information through.

Quick Set Up

Zero Assumptions can be installed onto your own company server in less than 5 minutes using our installation instructions.

After installation, the configuration file is created with all of the settings that the company will use. Alternatively, we offer many settings to be modified via the UI.

Configuration enables the control plane to start, and sets the following:

Server locations
User authentication
Email settings for password recovery and user invitations

Easy Ongoing Maintenance

Zero Assumptions ongoing maintenance is minimal.

Maintainers control the following: users, devices, groups, resources, nodes, and audit log access.

If maintainers want to go a step further, they can also automate the above controls.

Stay Connected to Zero Assumptions

Read more about Zero Assumptions.

Why Companies Need Audit Logs Now

Falon Darville — Mon, 10 Apr 2023 22:39:18 +0000

As we're doing more and more engineering work from different places, "the office" is no longer a static physical location. It could be home at the kitchen counter, at the airport, at a coffee shop, or even back in your childhood home. With this distribution of the workforce, companies have a duty to their stakeholders and to compliance protocols to keep audit logs for engineer activities.

In this article, we'll go over the ways in which Zero Assumptions provides security and data for unexpected activities performed by engineers.

Audit logs as insurance and assurance

Audit logs provide an insurance for tracking down errors and unexpected activities being performed on company databases and servers. When something unexpected happens, designated operators only need to log in to view audit logs and interpret the logs. From there, they can report out to stakeholders using real data rather than perceptions and assumptions.

Because Zero Assumptions is an on-prem solution, companies determine where to store logs and for how long. This means that companies have a high level of ownership over the compliance standards that they must uphold. Zero Assumptions gives that freedom back to companies, rather than dictating how data is stored.

With audit logs and control over data storage, companies can be assured that they have a solution in place in case of bad actors, breaches, and human error.

Bonus use cases for audit logs

Audit logs can do a lot more than provide a written record for investigation of unexpected activities. Here are some additional ideas for how to use audit log data:

Reveal inefficient database requests
Provide aggregation of activity for database and server cost analysis
Compile work hours to determine engineers' work habits (can help with exploring experimental programs like the 4-day work week, etc.)

Stay Connected to Zero Assumptions

Read more about Zero Assumptions.

Secure Data with On-Prem Storage of Engineer Activity

Falon Darville — Sun, 26 Mar 2023 00:06:09 +0000

Zero Assumptions offers companies a secure way to get and store their engineers’ activities on databases and servers. For companies, this means peace of mind over all actions performed on company information because there’s traceability.

Each action is attributed to a specific user device and location. In addition, we link time of action, access location, machine signature, and other defining metadata.

Zero Assumptions is low maintenance

Zero Assumptions doesn’t require significant staffing to roll out or upkeep. An IT role individual fills out the configuration file, and deploys the configuration onto an on-prem server.

Our product is easy to manage and maintain once it’s deployed. Ongoing management tasks include user management, device management, group management, resource management, and node management.

Zero Assumptions leverages audit logs

Activity is printed to an audit log that’s available at any time. Audit logs can also be exported to another location for storage purposes.

Audit logs can be used for internal and external auditing to verify engineer actions, and trace back root causes for issues that arise in a company's system.

Zero Assumptions helps companies be compliant

Companies have full control over their data, since Zero Assumptions doesn’t require cloud storage. On-prem works especially well if your company deals with regulated and sensitive data that you need to retain sole access to.

Zero Assumptions helps companies to be SOC2 and ISO 27001 compliant by providing tremendous flexibility.

Stay Connected to Zero Assumptions

Read more about Zero Assumptions.

Zero Assumption's Proxy Architecture

Falon Darville — Tue, 07 Feb 2023 02:02:25 +0000

Zero Assumptions (ZA) is a simple to configure proxy that's minimal invasive to engineers, and maximally beneficial to business. After on-prem installation, Zero Assumptions requires engineers to log in, and it monitors their activity within a private network - all in the background while they work. For businesses, there's no need to set up complex logins as ZA can leverage existing identity providers.

In this article, we'll go into Zero Assumption's logical proxy architecture. This architecture won't impact configuration, though it'll help businesses understand what's happening behind the scenes. We use a control plane and data plane separation, though ZA can be run from a single virtual machine. The two planes are mainly used for ease of maintenance and build.

Access our Design Principles documentation for more information.

Control Plane

The Zero Assumptions control plane can be considered the brain of the product. Here, information is inputted, retained, and used. The control plane holds definitions for identity management, including the business's preferred, ZA-compatible identity provider.

The control plane also holds ACLs (Access Control Lists), which define how engineers can interact with the software within scope of Zero Assumptions monitoring. Admins update ACLs through the configuration file or admin UI.

Now that we've mentioned configuration, there's a distinction between the configuration file written in UCL or JSON and data plane node configuration. The data plane node configuration defines how network traffic is routed, while the admin configuration defines how the product is configured for the engineer end users.

Data Plane

While the control plane handles the logic of the application, the data plane runs the overlay network's traffic. The data plane is low maintenance. Zero Assumptions has opted to keep it simple so businesses don't have to bother with continuously handling. Businesses using ZA are responsible for connecting the control plane to the data plane, and periodically
installing updates.

Altogether, Zero Assumption's proxy architecture is separated for the sake of keeping the codebase clean and easy to maintain on our end. Businesses benefit from this simplicity because they don't have to spend significant time learning how to use our product to get it up and running.

Stay Connected to Zero Assumptions

Read more about Zero Assumptions.

Zero Assumption’s Backbone of Rust

Falon Darville — Tue, 10 Jan 2023 00:57:38 +0000

Zero Assumptions is a zero trust networking solution you can use with your existing Identity Provider. All of your data is stored on-prem, so you control it all.

While you won't have to write any foundational code to drive the functionality of Zero Assumptions for your organization, this article explores our usage of the Rust language. The purpose being that you'll gain a stronger understanding of the mechanisms that drive Zero Assumptions, and why our product is secure, safe, and performant.

We've built our solution on Rust for a number of reasons including its strong community support, packages on crates.io, and future-proofing.

Decrease Runtime Errors

Given that Zero Assumptions is on-prem first. Given the nature of our product, if the software goes down, then people can’t access their private resources. We have to send product that is as solid as possible at runtime.

Rust is a memory-safe language that eliminates an entire class of errors without sacrificing performance. Specifically, in safe mode, Absence of Null prevents another class of bugs occurring in runtime.

Consistent Quality and Longevity

Rust contributors intentionally keep the language backwards compatible. This means that when Rust changes, Zero Assumptions won't introduce breaking changes due specifically to using a newer version of Rust.

When writing Rust itself, the language has no additional usage "cost" for abstractions. This will allow us to continue to iterate quickly without slowing down the end product.

It's also worth mentioning that Rust has robust and clear documentation, which greatly adds to the overall quality of the language.

Security

When Rust compiles, it looks at memory compilation, which can prevent memory management problems. Rust also allows for more granular control over memory, so we can highly customize where we want to.

Zero Assumptions benefits from its Rust foundation on many levels. And while our users don't need to know any Rust, they'll reap all the rewards of Zero Assumptions being built on Rust.

Follow Zero Assumptions here on DEV, LinkedIn, and Twitter to stay up-to-date on our product. We're launching this year, 2023, and can't wait for you to use Zero Assumptions.

Build a Job Queue with Rust Using Aide-De-Camp (Part 4)

Falon Darville — Tue, 13 Dec 2022 02:02:01 +0000

This is the fourth and final installment in our series. We continue to run you through creating durable job queues using Zero Assumption's open source solution, Aide-De-Camp, which is built using Rust.

In this post, we'll build cron job support into Aide-De-Camp.

Aide-De-Camp Code

Code from this series is available in our open source Aide-De-Camp GitHub repository.

The Crate is published to crates.io for everyone to use.

Challenges of Making a Cron Job Scheduler

Making a persistent and distributed cron job scheduler is easier said than done. There are many cases to account for that aren't immediately apparent. For example, in a distributed environment, there can only be one scheduler that keeps track of what has and hasn't been scheduled. Another example is the case for smart decision making. If you want to run a scheduler at midnight and the scheduler is off at midnight.

It's straightforward when you have a cluster with a leader with distributed storage. It's different when you're working with just a job queue and job processor.

Let's break down how a cron job scheduler functions.

From a high level, a cron job scheduler performs the following in a continuous loop.

Check if there is a job to run
If there is a job to run, run that job
Sleep for a designated period of time

Leader Election

If you have a single leader instance, this instance is always a leader because that's the only possible outcome of any election. What if you have two or more? Well, if instances don't coordinate between each other then your jobs will get scheduled N times every time, where N is number of schedulers.

One solution can be to have a single scheduler. However, the goal of Aide-De-Camp is to be "serverless" in that there isn't a special server running somewhere.

Aide-De-Camp won't handle the leader election process. This will be done by the crate consumer.

To Aide-De-Camp, the scheduler will take watch::Receiver<IsLeader> and check if the current instance is a leader before proceeding.

Cron Job Storage for Scheduling

We use cron job storage to keep track of what jobs we've scheduled and hold the following.

When a new job should be scheduled
When the last job was scheduled
Whether or not a job is already schedule during the requested job time

Storage enables us to time our cron jobs in a most sophisticated way than checking the schedule on a +/- 1 second cadence, or something similar. Scheduling a short cadenced check would saturate CPU if many tasks were running.

Let's look at in-memory storage for scheduling, which has some downsides.

With more than one scheduler, all schedulers must agree on what schedules exist
We can lose some ticks:
- Imagine there is a single scheduler
- There is a schedule that should fire at T
- Instance goes offline at T - x
- Instance comes back at T + y
- Scheduler calculated a new T that is the next tick
- This is fine if this task is scheduled often, but for another task, this will create issues

For Aide-De-Camp, we're going to implement in-memory store and SQLite store for now.

Another reason to have storage is to keep track what we've already scheduled. This is a bit tricker because we want to support multiple strategies to handle overlapping jobs:

Schedule anyway
Wait for current job to finish
Skip

Up Next

This post wraps up our first Aide-De-Camp series!

Aide-De-Camp is a product of Zero Assumptions.

Build a Job Queue with Rust Using Aide-De-Camp (Part 3)

Falon Darville — Tue, 06 Dec 2022 00:23:34 +0000

This is the third installment in our series, where we continue to run you through creating durable job queues using Zero Assumption's open source solution, Aide-De-Camp, which is built using Rust.

In this post, we'll go over feature requests we've received and how we created those features.

Aide-De-Camp Code

Code from this series is available in our open source Aide-De-Camp GitHub repository.

The Crate is published to crates.io for everyone to use.

Feature Requests

We'll address two feature requests we've implemented:

These two requests required changes to our public API, which created breaking changes.

To submit your own feature request for Aide-De-Camp, open an Issue on our open source GitHub repository.

Cancel Job By ID Feature

In this Issue, we were asked to implement cancelling a job by providing the job ID. See the Issue submitted on GitHub.

Cancel a Job That Hasn't Started

In our implementation, you can only cancel a job when it has not yet started.

A job ID is available once the job is submitted. We use this job ID value in a new Queue trait. In a method of the Queue trait, we do a couple things:

Check if the job has already started. If so, ignore the cancellation request.
Check if the job ID does not exist. If so, return an error indicating that the job can't be found.

If the job ID exists and the job has not started, the cancellation goes through. A payload for the cancelled job is returned when the job type is specified. If no job type is specified, then no payload returns but the job is still cancelled. You won't always be able to provide the job type as it's not always available.

In the following code, you'll see two cancel job functions:

Cancel scheduled job using job id, and return no payload upon success
Cancel scheduled job using job id and job type, and return payload upon success

/// Cancel job that has been scheduled. Right now this will only cancel if the job hasn't started yet.
async fn cancel_job(&self, job_id: Xid) -> Result<(), QueueError>;

/// The same as [`cancel_job`](struct.cancel_job.html), but returns payload of canceled job.
/// If deserialization fails, then job won't be cancelled.
async fn unschedule_job<J>(&self, job_id: Xid) -> Result<J::Payload, QueueError>
where
    J: JobProcessor + 'static,
    J::Payload: Decode;

To review the full job cancellation feature implementation, see Aide-De-Camp feature #9.

Unit Testing for Solution

With the above functionality, we tested against the following cases:

Job ID cannot be found
Job was cancelled successfully with payload returned since the job type was included in the cancellation request
Job was cancelled successfully without payload returned
job_type that does NOT match jid (for method that returns payload)
Cancelling the job with payload, but the payload couldn't be deserialized
Cannot cancel a job that has already started

Job Priority Feature

In this Issue, we were asked to provide a job prioritization ability. See the Issue submitted on GitHub.

Priority becomes important when there is worker exhaustion, or more jobs ready to be worked on than there are job processors. In that case, it's beneficial to run jobs with higher priority first even if they were submitted later than lower priority jobs.

We chose to implement priority using an integer, i8 or the range -128 to 127. The higher the number, the higher the priority.

Given this, the queue interface needed to be changed. We added a required argument priority: <i8 value>. We recommend setting low priority jobs to priority: 0.

A SQL query looks at jobs in the queue and runs through the priority values, ordering them from highest number to lowest number.

SELECT jid FROM adc_queue WHERE started_at IS NULL AND scheduled_at <= now() AND job_type IN (“my_job”) ORDER BY priority DESC LIMIT 1

To review the full job prioritization feature implementation, see Aide-De-Camp feature #10.

Up Next

In this post, we explained two feature requests, job cancellation and job prioritization, and how we implemented them. In part 4, we'll go over building cron job support.

Aide-De-Camp is a product of Zero Assumptions.

Simplify Auditing with Zero Assumptions

Falon Darville — Tue, 29 Nov 2022 03:33:14 +0000

Zero Assumptions is a simple-to-configure access control that monitors engineers' work. Use your own Identity Provider (IDP) and build out your highly customizable configuration using our uncomplicated parameters.

For engineers, Zero Assumptions is easy to use and low effort. They log into the IDP and connect to the desired resource. No slowdown for them, and no assumptions for you.

Zero Assumptions purposefully enables organizations to create audit trails. It covers the missing authentication, authorization, and audit pieces. We provide the information that administrators need to hold their users accountable to their actions.

Password Deep Dive

Password sharing means that multiple individuals use the same password to authenticate and perform tasks. It's bad practice. Let's get into some specific reasons why.

The practice of sharing passwords poses many issues and complications, including the following:

Rotating shared passwords is a nightmare. There's no certainty around who is using the password, so some individuals may need to be tracked down (to share the new password with). Some individuals may get locked out and passwords may be shared over unreliable networks, opening the password up to being used by unintended individuals.
Every time an individual off-boards or needs to lose access to an account, someone has to remember to change the password. It's easy to forget and leave the account open to unwanted access.
It's easy to use a password manager to store and populate website passwords. It's more convoluted to use a password manager for databases and servers.
Auditing actions on a shared password account is hard, and sometimes nearly impossible.
If a password is leaked and precautions are taken to regain the account, then all users will be locked out.

To prevent getting into the password trap, we recommend:

Don't wait until password sharing becomes a problem. Access your situation early to come up with an alternative.
When determining if a solution is right for you, imagine it in scale. What's it going to be like to manage the solution for an exponential amount more of resources and people?
Understand that more passwords is not the solution.

At the end of the day, the purpose of a password is to provide access and associate that access with an identity. With Zero Assumptions, we're particularly interested in the auditing aspect, which is why shared passwords are never in the equation. Let's look at how we associate individuals to their actions.

Auditing and Individual Identities

With shared passwords, audit trails cannot be trusted. Multiple people use the same credentials, so it's hard to tell who is doing what. Who's running a very long query in production? Who downloaded all of your customer's social security numbers on a Friday night?

You need to be able to authenticate for accountability, and you can't do this without uniquely authenticating all users and/or all user devices.

Being unable to identify who's performing certain actions also won't pass certifications like SOC Type-2 or PCI-DSS.

With Zero Assumptions, you're part way to fulfilling certification criteria since you're accounting for self-managed off-boarding.

Reduce Password Usage

Zero Assumptions lets you bring your own IDP: Google Workspaces, Okta, and generic OIDC. With an IDP, you don't have to manage many users using many services.

Your users won't have to remember yet another password, and they'll be individually identified. All their actions will be directly associated with them.

Stay Connected to Zero Assumptions

Read more about Zero Assumptions.

Build a Job Queue with Rust Using Aide-De-Camp (Part 2)

Falon Darville — Thu, 17 Nov 2022 03:00:06 +0000

This is the second installment in our series, which runs you through creating durable job queues using Zero Assumption's open source Rust solution, Aide-De-Camp.

In part one of our series, we went over the Rust traits that we created for the job queue. In this post, we're implementing the traits we talked about in part one.

Aide-De-Camp Code

Code from this series is available in our open source Aide-De-Camp GitHub repository.

The Crate is published to crates.io for everyone to use.

SQLite Schema

SQLite is just the first backend we've implemented. More backend implementations to come.

Aide-De-Camp uses SQLite to persist the queue. Let's look at the schema.

Jobs go into the main queue, adc_queue. Dead letter jobs, or jobs that couldn't be completed, go into the adc_dead_queue, where you can investigate them later.

CREATE TABLE IF NOT EXISTS adc_queue (
 jid TEXT PRIMARY KEY,
 queue TEXT NOT NULL default 'default',
 job_type TEXT not null,
 payload blob not null,
 retries int not null default 0,
 scheduled_at INTEGER not null,
 started_at INTEGER,
 enqueued_at INTEGER not null default (strftime('%s', 'now'))
);

CREATE TABLE IF NOT EXISTS adc_dead_queue (
 jid TEXT PRIMARY KEY,
 queue TEXT NOT NULL,
 job_type TEXT not null,
 payload blob not null,
 retries int not null,
 scheduled_at INTEGER not null,
 started_at INTEGER not null,
 enqueued_at INTEGER not null,
 died_at INTEGER not null default (strftime('%s', 'now'))
);

CREATE INDEX IF NOT EXISTS adc_queue_jobs ON adc_queue (
    scheduled_at asc,
    started_at asc,
    queue,
    job_type
);

Queue Implementation

Next, we implement the queue using Rust. The following code defines the connection pool and bincode configuration.

We only need to implement two methods, scheduled_at and poll_next_with_instant.

#[derive(Clone)]
pub struct SqliteQueue {
    pool: SqlitePool,
    bincode_config: bincode::config::Configuration,
}

scheduled_at

To schedule a job, all we need to do is serialize the payload, create a new JID (we use XID for JIDs), and insert both into the adc_queue table.

    #[instrument(skip_all, err, ret, fields(job_type = J::name(), payload_size))]
    async fn schedule_at<J>(
        &self,
        payload: J::Payload,
        scheduled_at: DateTime,
    ) -> Result<Xid, QueueError>
    where
        J: JobProcessor + 'static,
        J::Payload: Encode,
    {
        let payload = bincode::encode_to_vec(&payload, self.bincode_config)?;
        let jid = new_xid();
        let jid_string = jid.to_string();
        let job_type = J::name();

        tracing::Span::current().record("payload_size", &payload.len());

        sqlx::query!(
            "INSERT INTO adc_queue (jid,job_type,payload,scheduled_at) VALUES (?1,?2,?3,?4)",
            jid_string,
            job_type,
            payload,
            scheduled_at
        )
        .execute(&self.pool)
        .await
        .context("Failed to add job to the queue")?;
        Ok(jid)
    }

poll_next_with_instant

Since SQLite doesn't have built-in notifications for new jobs queued, we poll the queue periodically. In our UPDATE statement, we mark the row as in progress, then use the RETURNING * clause to retrieve job metadata.

The following is an excerpt from our 0.1.0 tag queue.rs source code. In our next posts, we'll review the modifications we've made to poll_next_with_instant.

UPDATE adc_queue SET started_at=?, retries=retries+1 WHERE jid IN (SELECT job from adc_queue WHERE started_at IS NULL AND queue=? AND scheduled_at <= ? AND job_type IN (?,?) LIMIT 1) RETURNING *;

JobHandle Trait Implementation

Since the entire job lifecycle is within the JobHandle trait, we have to implement that trait too.

pub struct SqliteJobHandle {
    pool: SqlitePool,
    row: JobRow,
}

#[async_trait]
impl JobHandle for SqliteJobHandle {
    fn id(&self) -> Xid {
        self.row.jid
    }

    fn job_type(&self) -> &str {
        &self.row.job_type
    }

    fn payload(&self) -> Bytes {
        self.row.payload.clone()
    }

    fn retries(&self) -> u32 {
        self.row.retries
    }

    async fn complete(mut self) -> Result<(), QueueError> {
        let jid = self.row.jid.to_string();
        sqlx::query!("DELETE FROM adc_queue where jid = ?1", jid)
            .execute(&self.pool)
            .await
            .context("Failed to mark job as completed")?;
        Ok(())
    }

    async fn fail(mut self) -> Result<(), QueueError> {
        let jid = self.row.jid.to_string();
        sqlx::query!("UPDATE adc_queue SET started_at=null WHERE jid = ?1", jid)
            .execute(&self.pool)
            .await
            .context("Failed to mark job as failed")?;
        Ok(())
    }

    async fn dead_queue(mut self) -> Result<(), QueueError> {
        let jid = self.row.jid.to_string();
        let retries = self.row.retries;
        let job_type = self.row.job_type.clone();
        let payload = self.row.payload.as_ref();
        let scheduled_at = self.row.scheduled_at;
        let enqueued_at = self.row.enqueued_at;

        let mut tx = self
            .pool
            .begin()
            .await
            .context("Failed to start transaction")?;
        sqlx::query!("DELETE FROM adc_queue WHERE jid = ?1", jid)
            .execute(&mut tx)
            .await
            .context("Failed to delete job from the queue")?;

        sqlx::query!("INSERT INTO adc_dead_queue (jid, job_type, payload, retries, scheduled_at, enqueued_at) VALUES (?1, ?2, ?3, ?4,?5,?6)",
            jid,
            job_type,
            payload,
            retries,
            scheduled_at,
            enqueued_at)
            .execute(&mut tx).await
            .context("Failed to move job to dead queue")?;
        Ok(())
    }
}

Up Next

So far, we've gone over creating traits and implementing them. In parts 3 and 4, we'll review how we modified our code based on community feature requests.

Aide-De-Camp is a product of Zero Assumptions.

Build a Job Queue with Rust Using Aide-De-Camp (Part 1)

Falon Darville — Sun, 30 Oct 2022 01:19:39 +0000

This is the first installment in a series that runs you through creating durable job queues using Zero Assumption's open source solution, Aide-De-Camp, which is built using Rust. For full transparency, hi - I'm Zero Assumption's creator, Andrey Snow. Follow along as I make it easier for you to manage job queueing using Rust.

Aide-De-Camp Code

Code from this series is available in our open source Aide-De-Camp GitHub repository.

The Crate is published to crates.io for everyone to use.

Intro to Queues

From time to time, your application needs to offload some work to a different process or delay execution of a certain task. This is where job queues come in handy.

Many existing job queues are opinionated on how to store and process jobs. For example, many require Redis running, some only work with PostgreSQL, and others require some external manager process running. It's not a problem whether you're already using Redis or PostgreSQL, but what if you want to use something lighter like SQLite and run a single process?

With very little code, SQLite and PostgreSQL can be turned into a durable job queue.

First, let's define our interfaces. Traits allow us to make our job queue backend agnostic. Maybe today using SQLite, tomorrow using PostgreSQL, and the day after using Redis.

Job

Job Lifecycle

Before we begin, let's define the expected job behavior. A job is any distinct work to be done. Even the same job task repeated will have a unique job ID.

We're going to visualize job lifecycle from the runner point of view as a Finite-state machine.

The above lifecycle has two kinds of errors, JobError and RunnerError.

JobError is an error specific to processing the job itself, where there's a problem decoding the input.

Errors with the handler are passed to the RunnerRouter. Handler-specific errors must implement the Into<JobError> trait. The easiest way to handle errors is to wrap your error in anyhow::Error.

RunnerError are queue domain errors and indicate a runtime issue. A runner error is thrown in the case of failure to transition a job due to an underlying QueueError, given a job that can't be handled by the current RunnerRouter. These errors are bubbled up and result in temporary suspension of a worker in hope that issue will go away in the future.

`JobProcessor` Trait

Use the JobProcessor trait to implement the Aide-De-Camp library. Implementation looks like the following.



#[async_trait]
pub trait JobProcessor: Send + Sync {
    /// What is the input to this handler. If you want to use `RunnerRouter`, then this must implement `bincode::Decode` and `bincode::Encode`.
    type Payload: Send;
    /// Which error is returned
    type Error: Send;
    /// Run the job, passing the payload to it. Your payload should implement `bincode::Decode`.
    async fn handle(&self, jid: Xid, payload: Self::Payload) -> Result<(), Self::Error>;

    /// How many times job should be re-tried before being moved to dead queue
    fn max_retries(&self) -> u32 {
        0
    }

    /// Job type, used to differentiate between different jobs in the queue.
    fn name() -> &'static str
    where
        Self: Sized;
}

A job needs the following.

Include the Payload type of Send. For example, type Payload: Send.
Declare an error type, such as type Error: Send. The error type can also be Infallible.
run function that takes jid (job_id shortened) and its input (payload) as arguments. For example, async fn handle(&self, jid: Xid, payload: Self::Payload) -> Result<(), Self::Error>;.
name function that returns the job name. This is used to distinguish jobs in the queue and used in RunnerRouter. For example, fn name() -> &'static str.
Be Send + Sync.
- Optionally, specify how many times a job should be re-tried upon failure. The default value is 1, meaning the job only has one chance to succeed.

Some queue implementations also pass Context to the job, and Context acts like an Extension of Axum. In this queue implementation, if your job has external dependencies (for example, an API client for your transactional email service provider), then it should be part of an entity that implements the Job trait. This way, you choose how to wire the dependencies, whether by passing them manually or using the Dependency Injection framework.

Type-Erased Job Handler

Rust has some limitations on the usage of the following: HashMap<String, JobType1|JobType2>.

To work around this, many Rust queues either:

Make your job handler accept some form of &[u8] as input, which requires too much boilerplate code.
Declare an enum with all possible job types, which limits flexibility. The publisher needs to be aware of every possible job, and it's hard to separate which jobs run where.

axum provides a workaround. After an hour of "Go-to-definition" between multiple tokio crates, I've come up with the following.



/// Object-safe implementation of a job that can be used in the runner
pub struct WrappedJobHandler<T: JobProcessor> {
    job: T,
    config: Configuration,
}

impl<J> WrappedJobHandler<J>
where
    J: JobProcessor + 'static,
    J::Payload: Decode + Encode,
    J::Error: Into<JobError>,
{
    pub fn new(job: J) -> Self {
        let config = bincode::config::standard();
        Self { job, config }
    }

    pub fn boxed(self) -> BoxedJobHandler {
        Box::new(self) as BoxedJobHandler
    }
}

#[async_trait::async_trait]
impl<J> JobProcessor for WrappedJobHandler<J>
where
    J: JobProcessor + 'static,
    J::Payload: Decode + Encode,
    J::Error: Into<JobError>,
{
    type Payload = Bytes;
    type Error = JobError;

    async fn handle(&self, jid: Xid, payload: Self::Payload) -> Result<(), Self::Error> {
        let (payload, _) = bincode::decode_from_slice(payload.as_ref(), self.config)?;
        self.job.handle(jid, payload).await.map_err(Into::into)
    }

    fn name() -> &'static str {
        J::name()
    }
}

Now, our RunnerRouter just needs to know about Box<dyn Job<Payload=Bytes, Error=JobError>. WrappedJobHandler<J> will handle decoding the payload and errors.

Handle Multiple Job Types in a Single Runner

I've already mentioned RunnerRouter multiple times without explaining what it is. You can think about it as an HTTP request router. I was seriously considering using router from axum underneath and make everything that implements Job also implement Tower's Service, but it seemed a little too much, so I've made my own:



#[derive(Default)]
pub struct RunnerRouter {
    jobs: HashMap<&'static str, BoxedJobHandler>,
}

impl RunnerRouter {
    /// Register a job handler with the router. If job by that name already present, it will get replaced.
    pub fn add_job_handler<J>(&mut self, job: J)
    where
        J: JobProcessor + 'static,
        J::Payload: Decode + Encode,
        J::Error: Into<JobError>,
    {
        let name = J::name();
        let boxed = WrappedJobHandler::new(job).boxed();
        self.jobs.entry(name).or_insert(boxed);
    }

    pub fn types(&self) -> Vec<&'static str> {
        self.jobs.keys().copied().collect()
    }

    /// Process job handle. This function is responsible for job lifecycle. If you're implementing your
    /// own job runner, then this is what you should use to process a job that is already pulled
    /// from the queue. In all other cases, you shouldn't use this function directly.
    #[instrument(skip_all, err, fields(job_type = %job_handle.job_type(), jid = %job_handle.id().to_string(), retries = job_handle.retries()))]
    pub async fn process<H: JobHandle>(&self, job_handle: H) -> Result<(), RunnerError> {
        if let Some(r) = self.jobs.get(job_handle.job_type()) {
            match r
                .handle(job_handle.id(), job_handle.payload())
                .await
                .map_err(JobError::from)
            {
                Ok(_) => {
                    job_handle.complete().await?;
                    Ok(())
                }
                Err(e) => {
                    tracing::error!("Error during job processing: {}", e);
                    if job_handle.retries() >= r.max_retries() {
                        tracing::warn!("Moving job {} to dead queue", job_handle.id().to_string());
                        job_handle.dead_queue().await?;
                        Ok(())
                    } else {
                        job_handle.fail().await?;
                        Ok(())
                    }
                }
            }
        } else {
            Err(RunnerError::UnknownJobType(
                job_handle.job_type().to_string(),
            ))
        }
    }

}

This isn't a complete version of it, we will expand it later, but for now, that's enough. Don't worry about JobHandle trait just yet, I will get to it soon.

Registering job handlers

Handlers can only be registered while you have a mutable reference to RunnerRouter. I've done it like that to avoid having to lock jobs behind RwLock or making a separate builder structure. Intended use:



// create your job handler here
let my_job = MyJob::default();
let router = {
    let mut r = RunnerRouter::default();
    r.add_job_handler(my_job);
    r
};

Job processing

This is handled by process function. It takes a type that implements JobProcessor trait as input. This method is in charge of handling the entire job lifecycle.

First, we check if the current RunnerRouter is capable of handling the given job. Then, we attempt to process the given job. Finally, we have the option to do the following:

Remove the job from the queue
Re-queue the job
Move the job to the dead queue if the retry count has exceeded the desired maximum

Queue

Until now, we didn't talk about the queue itself. Abstracted queue works like this:



while let Some(job) = queue.pop() {
    job.process();
}

This wouldn't be a very efficient queue for many reasons. Our queue is going to look more like this:

Queue trait

Now that we have the idea of how queue works, we can define its trait:



#[async_trait]
pub trait Queue: Send + Sync {
    type JobHandle: JobHandle;
    /// Schedule a job to run at the future time.
    async fn schedule_at<J>(
        &self,
        payload: J::Payload,
        scheduled_at: DateTime,
    ) -> Result<Xid, QueueError>
    where
        J: JobProcessor + 'static,
        J::Payload: Encode;
    /// Schedule a job to run next. Depending on queue backlog this may start running later than you expect.
    async fn schedule<J>(&self, payload: J::Payload) -> Result<Xid, QueueError>
    where
        J: JobProcessor + 'static,
        J::Payload: Encode,
    {
        self.schedule_at::<J>(payload, Utc::now()).await
    }

    /// Schedule a job to run at the future time relative to now.
    async fn schedule_in<J>(
        &self,
        payload: J::Payload,
        scheduled_in: Duration,
    ) -> Result<Xid, QueueError>
    where
        J: JobProcessor + 'static,
        J::Payload: Encode,
    {
        let when = Utc::now() + scheduled_in;
        self.schedule_at::<J>(payload, when).await
    }

    /// Pool queue, implementation should not wait for next job, if there nothing return `Ok(None)`.
    async fn poll_next_with_instant(
        &self,
        job_types: &[&str],
        time: DateTime,
    ) -> Result<Option<Self::JobHandle>, QueueError>;

    /// Pool queue, implementation should not wait for next job, if there nothing return `Ok(None)`.
    async fn poll_next(&self, job_types: &[&str]) -> Result<Option<Self::JobHandle>, QueueError> {
        self.poll_next_with_instant(job_types, Utc::now()).await
    }

    /// Await next job. Default implementation polls the queue with defined interval until there is something.
    async fn next(
        &self,
        job_types: &[&str],
        interval: Duration,
    ) -> Result<Self::JobHandle, QueueError> {
        let duration = interval
            .to_std()
            .map_err(|_| QueueError::InvalidInterval(interval))?;
        let mut interval = tokio::time::interval(duration);
        loop {
            interval.tick().await;
            let job = self.poll_next(job_types).await?;
            if let Some(job) = job {
                break Ok(job);
            }
        }
    }
}

This trait designed to work only with jobs that can be handled by RunnerRouter. This trait provides convenient method to scheduling jobs: schedule for scheduling job right now, schedule_in for scheduling job in the future (e.g., in 5 minutes), schedule_at for scheduling job at the specific time (e.g., Mon, 16 May 2022 21:02:35 GMT). Methods that implementation needs to implement are: schedule_at and poll_next_with_instant.

schedule_at is self-explanatory: these methods adds a job to the queue to run at a specific time.

poll_next_with_instant is a bit more complicated. It takes a list of job types and time as arguments. time is used to determine what should be treated as "now" when queue polled. Using time as an input is very helpful because it allows time travel in tests. The implementation should check the queue and return a JobHandle to a job if there is anything in the queue.

There is also next function that in a loop polls the queue, if it's empty, it sleeps a specified interval and polls again until there is something. This is just a default implementation, more specific implementation can use combination of sleeping and external wake-ups (e.g. NOTIFY from PostgreSQL or SUBSCRIBE from Redis).

`JobHandle`

You might have already noticed that most of the job lifecycle handled by some JobHandle. What is it? Well…it's an implementation specific way of handling job lifecycle. Queue only two jobs: sending jobs and polling for jobs. Queue here is an interface to a queue that exists somewhere, rather the queue itself.

JobHandle trait looks like this:



#[async_trait]
pub trait JobHandle: Send + Sync {
    // Get job id
    fn id(&self) -> Xid;
    // Get Job type
    fn job_type(&self) -> &str;
    // Get job payload.
    fn payload(&self) -> Bytes;
    // How many times this job has been retried already.
    fn retries(&self) -> u32;
    // Mark the job as completed successfully.
    async fn complete(mut self) -> Result<(), QueueError>;
    // Mark the job as failed
    async fn fail(mut self) -> Result<(), QueueError>;
    // Move the job to dead queue.
    async fn dead_queue(mut self) -> Result<(), QueueError>;
}

Everything past "job checked out of the queue" in job lifecycle is handled within its implementation.

Processing the Queue

Extending `RunnerRouter`

Now that we have an interface to interact with our queue, we can extend RunnerRouter to work with it:



impl RunnerRouter {
    pub async fn listen<Q, QR>(&self, queue: Q, poll_interval: Duration)
    where
        Q: AsRef<QR>,
        QR: Queue,
    {
        let job_types = self.types();
        loop {
            match queue.as_ref().next(&job_types, poll_interval).await {
                Ok(handle) => match self.process(handle).await {
                    Ok(_) => {}
                    Err(RunnerError::QueueError(e)) => handle_queue_error(e).await,
                    Err(RunnerError::UnknownJobType(name)) => {
                        tracing::error!("Unknown job type: {}", name)
                    }
                },
                Err(e) => {
                    handle_queue_error(e).await;
                }
            }
        }
    }
}
async fn handle_queue_error(error: QueueError) {
    tracing::error!("Encountered QueueError: {}", error);
    tracing::warn!("Suspending worker for 5 seconds");
    tokio::time::sleep(std::time::Duration::from_secs(5)).await;
}

In a loop, we poll the queue and handle the incoming jobs. If we encounter a QueueError during polling, then worker goes into a short sleep. If we encounter a RunnerError, then we just log it and work on the next job.

Putting it all together

Technically, we can just create a queue, pass it to RunnerRouter::listen, and call it a day. The only issue is that it's processing a single job at a time. In some cases, that's enough. But we want to go fast.

Full disclaimer - the following is a naive version of a job runner. It works, but has some limitations.



const JITTER_INTERVAL_MS: [i64; 10] = [0, 1, 1, 2, 3, 5, 8, 13, 21, 34];

pub struct JobRunner<Q>
where
    Q: Queue,
{
    queue: Arc<Q>,
    processor: Arc<RunnerRouter>,
    semaphore: Arc<Semaphore>,
}

impl<Q> JobRunner<Q>
where
    Q: Queue + 'static,
{
    pub fn new(queue: Q, processor: RunnerRouter, concurrency: usize) -> Self {
        Self {
            queue: Arc::new(queue),
            processor: Arc::new(processor),
            semaphore: Arc::new(Semaphore::new(concurrency)),
        }
    }
}

impl<Q> JobRunner<Q>
where
    Q: Queue + 'static,
{
    pub async fn run(&mut self, interval: chrono::Duration) -> Result<(), QueueError> {
        loop {
            let semaphore = self.semaphore.clone();
            let permit = semaphore
                .acquire_owned()
                .await
                .context("Semaphore closed while running")?;
            let queue = self.queue.clone();
            let processor = self.processor.clone();
            tokio::spawn(async move {
                let _permit = permit;
                let queue = queue;
                let processor = processor;
                let interval = interval + get_random_jitter();
                processor.listen(queue, interval).await;
            });
        }
    }
}

fn get_random_jitter() -> chrono::Duration {
    JITTER_INTERVAL_MS
        .choose(&mut rand::thread_rng())
        .map(|ms| chrono::Duration::milliseconds(*ms))
        .unwrap_or_else(|| chrono::Duration::milliseconds(5))
}

First, let's talk about the JobRunner:



pub struct JobRunner<Q>
where
    Q: Queue,
{
    queue: Arc<Q>,
    processor: Arc<RunnerRouter>,
    semaphore: Arc<Semaphore>,
}

impl<Q> JobRunner<Q>
where
    Q: Queue + 'static,
{
    pub fn new(queue: Q, processor: RunnerRouter, concurrency: usize) -> Self {
        Self {
            queue: Arc::new(queue),
            processor: Arc::new(processor),
            semaphore: Arc::new(Semaphore::new(concurrency)),
        }
    }
}

We already know why we need the first two, but what about Semaphore? Semaphore
is a synchronization primitive that limits concurrent access to a resource.

Essentially, it's a metered highway ramp for your program. We're going to utilize it to limit how many jobs could be in flight at any given moment.

Generally, it's a bad idea to have N things hammer a resource at the same time, so we're going to introduce a random-ish jitter:



const JITTER_INTERVAL_MS: [i64; 10] = [0, 1, 1, 2, 3, 5, 8, 13, 21, 34];

fn get_random_jitter() -> chrono::Duration {
    JITTER_INTERVAL_MS
        .choose(&mut rand::thread_rng())
        .map(|ms| chrono::Duration::milliseconds(*ms))
        .unwrap_or_else(|| chrono::Duration::milliseconds(5))
}

It doesn't have to be a Fibonacci series, but using this series allows for some pretty unique patterns.

Now, the job begins processing. First, we start a loop. Then we acquire a "permit" from our semaphore:



  let semaphore = self.semaphore.clone();
  let permit = semaphore
      .acquire_owned()
      .await
      .context("Semaphore closed while running")?;

The await will pause the loop until the permit becomes available. After, we clone our Queue and RunnerRouter:



  let queue = self.queue.clone();
  let processor = self.processor.clone();

Finally, we start polling our queue and process incoming job:



  tokio::spawn(async move {
      let _permit = permit;
      let queue = queue;
      let processor = processor;
      let interval = interval + get_random_jitter();
      processor.listen(queue, interval).await;
  });

We spawn a new future into the tokio reactor and that's it. The rest is implemented in RunnerRouter. When the job completes, the permit will return to our semaphore and the cycle repeats.

Possible Improvements

Dynamic concurrency

Currently, concurrency is set in stone and can't be changed after JobRunner started its work. That means with concurrency of 50, there will be 50 tasks constantly asking a queue for new jobs. I want concurrency to shrink and grow depending on the current load.

Shrinking and growing concurrency can be done by ballooning semaphore permits. A task can check queue size and acquire or release permits based on that. It's a bit harder to reclaim permits that are already taken by a waiting worker, but it is possible and may require changing the interface of Queue trait.

Hooks

There is very little insight into what is happening in the queue scheduler and worker process. In the future, I''ll add a way to get notified when certain events happen. Hooks be used to drive dynamic concurrency.

Batching

Instead of polling jobs one by one, I want to poll N jobs at once. Batching is helpful for a situation where we have many small jobs constantly sent to the queue.

Multiple Queues

Currently, jobs are split by type and that's the only way to have different concurrency and/or poll interval. This is fine for many use cases, but occasionally, we want to have "hot" and "cold" queues that can have the same types of jobs. For example, you're working on a SaaS product with a free trial where tasks from free tier clients can go to a slow queue, while paid clients go to fast queue.

Queue Insights

Not being able to see queue state isn't a big deal if your jobs never fail (spoiler alert: jobs do fail). Our Queue trait can be extended to provide some insight. Namely, how many jobs in the queue, how many jobs should be executed right now, and what is going on in the dead queue.

Up Next

So far, we only covered generic items like traits and glue between generic components. Part 2 is going to cover SQLite implementation of Queue trait and benchmarking.

Aide-De-Camp is a product of Zero Assumptions.

DEV Community: Falon Darville

Simple IT: Set Up and Ongoing Maintenance of Zero Assumptions

Quick Set Up

Easy Ongoing Maintenance

Stay Connected to Zero Assumptions

Why Companies Need Audit Logs Now

Audit logs as insurance and assurance

Bonus use cases for audit logs

Stay Connected to Zero Assumptions

Secure Data with On-Prem Storage of Engineer Activity

Zero Assumptions is low maintenance

Zero Assumptions leverages audit logs

Zero Assumptions helps companies be compliant

Stay Connected to Zero Assumptions

Zero Assumption's Proxy Architecture

Control Plane

Data Plane

Stay Connected to Zero Assumptions

Zero Assumption’s Backbone of Rust

Decrease Runtime Errors

Consistent Quality and Longevity

Security

Build a Job Queue with Rust Using Aide-De-Camp (Part 4)

Aide-De-Camp Code

Challenges of Making a Cron Job Scheduler

Leader Election

Cron Job Storage for Scheduling

Up Next

Build a Job Queue with Rust Using Aide-De-Camp (Part 3)

Aide-De-Camp Code

Feature Requests

Cancel Job By ID Feature

Cancel a Job That Hasn't Started

Unit Testing for Solution

Job Priority Feature

Up Next

Simplify Auditing with Zero Assumptions

Password Deep Dive

Auditing and Individual Identities

Reduce Password Usage

Stay Connected to Zero Assumptions

Build a Job Queue with Rust Using Aide-De-Camp (Part 2)

Aide-De-Camp Code

SQLite Schema

Queue Implementation

scheduled_at

poll_next_with_instant

JobHandle Trait Implementation

Up Next

Build a Job Queue with Rust Using Aide-De-Camp (Part 1)

Aide-De-Camp Code

Intro to Queues

Job

Job Lifecycle

JobProcessor Trait

Type-Erased Job Handler

Handle Multiple Job Types in a Single Runner

Registering job handlers

Job processing

Queue

Queue trait

JobHandle

Processing the Queue

Extending RunnerRouter

Putting it all together

Possible Improvements

Dynamic concurrency

Hooks

Batching

Multiple Queues

Queue Insights

Up Next

`JobProcessor` Trait

`JobHandle`

Extending `RunnerRouter`