Job Queues

.travis.yml

@@ -6,14 +6,13 @@ rust:
 addons:
  config:
  retries: true
+ postgresql: 9.6
 
 cache:
  cargo: true
  directories:
  - /home/travis/.rvm
 
-services:
- - postgresql
 
 before_install:
  - sudo apt-get update -yqq

Cargo.toml

@@ -11,47 +11,50 @@ default-run = "rustodon"
 members = ['lib/resopt/', 'lib/posticle/']
 
 [dependencies]
-dotenv = "0.13"
-maplit = "1.0"
+dotenv = "0.13.0"
+maplit = "1.0.1"
 failure = "0.1.5"
 failure_derive = "0.1.5"
-lazy_static = "1.3"
+lazy_static = "1.3.0"
 itertools = "0.8.0"
 resopt = { path = "lib/resopt/" }
-structopt = "0.2.14"
+structopt = "0.2.15"
+prettytable-rs = { version = "0.8", default-features = false, features = ["win_crlf"] }
 
-regex = "1.1.0"
+regex = "1.1.2"
 posticle = { path = "lib/posticle/" }
 
-slog = "2.4"
-slog-term = "2.4"
-slog-async = "2.3"
-slog-scope = "4.0"
-rocket-slog = "0.4"
+slog = "2.4.1"
+slog-term = "2.4.0"
+slog-async = "2.3.0"
+slog-scope = "4.1.1"
+rocket-slog = "0.4.0"
 
-rocket = "0.4"
+rocket = "0.4.0"
 
-serde = "1.0"
-serde_derive = "1.0"
-serde_json = "1.0"
-validator = "0.8"
-validator_derive = "0.8"
+serde = "1.0.89"
+serde_derive = "1.0.89"
+serde_json = "1.0.39"
+validator = "0.8.0"
+validator_derive = "0.8.0"
 
 askama = { version = "0.8.0", features = ["with-rocket"] }
 ammonia = "2.0.0"
 
-diesel = { version = "1.4.1", features = ["postgres", "chrono", "r2d2"] }
+diesel = { version = "1.4.1", features = ["postgres", "chrono", "serde_json", "r2d2"] }
+diesel-derive-enum = { version = "0.4.4", features = ["postgres"] }
 flaken = "0.2.2"
+turnstile = { path = "lib/turnstile/" }
 
-chrono = "0.4"
+chrono = "0.4.6"
 chrono-humanize = "0.0.11"
 
-openssl = "0.10.16"
-pwhash = "0.3"
+openssl = "0.10.19"
+pwhash = "0.3.0"
 
-base32 = "0.4"
+base32 = "0.4.0"
 
 [dependencies.rocket_contrib]
-version = "0.4"
+version = "0.4.0"
 default-features = false
 features = ["json"]

diesel.toml

@@ -1,3 +1,4 @@
 [print_schema]
 file = "src/db/schema.rs"
 with_docs = true
+import_types = ["diesel::sql_types::*", "crate::db::types::Job_status"]

docs/internals/workers.md

@@ -0,0 +1,16 @@
+okay so this is how jobs are gonna work
+* spawn "dispatch" thread
+ * this boy just sits there in a nice little loop {}
+ * tries to grab N jobs from the db
+ * if it can't, just yields its time slice
+ * figure out what jobs we should run _now_ (`can_run: Vec<&dyn Job>`)
+ * ```sql
+ UPDATE jobs SET jobstatus = INFLIGHT, updated = #{time.now} WHERE id IN can_run
+ ```
+ * push into thread queue
+* spawn "timeout" thread
+ * also doing a loop {}
+ * ```sql
+ SELECT jobid FROM jobs WHERE jobstatus = INFLIGHT, "updated" + "timeout" < #{time.now}
+ ```
+ * do we need to kill inflight jobs?

lib/turnstile/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "turnstile"
+version = "0.1.0"
+edition = "2018"
+authors = ["Erin Moon <[email protected]>"]
+
+[dependencies]
+threadpool = "1.7"
+serde = "1.0"
+serde_json = "1.0"
+failure = "0.1.5"

lib/turnstile/README.md

@@ -0,0 +1,5 @@
+# turnstile
+thanks for the name, tef and cosine (っ´ω｀)っ
+
+## what's this?
+turnstile is a crate that wraps up a bunch of the types we use for job scheduling in one nice package.

lib/turnstile/src/error.rs

@@ -0,0 +1,46 @@
+use failure::Fail;
+use serde_json;
+use std::any::Any;
+use std::fmt::{self, Debug, Display};
+use std::sync::Mutex;
+
+#[derive(Fail, Debug)]
+pub enum Error {
+ #[fail(display = "Inner error in job: {}", _0)]
+ JobInnerError(#[fail(cause)] failure::Error),
+
+ #[fail(display = "Job panicked: {:?}", _0)]
+ JobPanicked(#[fail(cause)] SyncPanicError),
+
+ #[fail(display = "Error deserializing job data: {}", _0)]
+ DeserializeError(#[fail(cause)] serde_json::Error),
+
+ #[fail(display = "Invalid kind for job")]
+ InvalidKind,
+}
+
+pub struct SyncPanicError {
+ inner: Mutex<Box<dyn Any + Send + 'static>>,
+}
+
+impl SyncPanicError {
+ pub(crate) fn new(inner: Box<dyn Any + Send + 'static>) -> Self {
+ Self {
+ inner: Mutex::new(inner),
+ }
+ }
+}
+
+impl Display for SyncPanicError {
+ fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+ (*self.inner.lock().unwrap()).fmt(f)
+ }
+}
+
+impl Debug for SyncPanicError {
+ fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+ (*self.inner.lock().unwrap()).fmt(f)
+ }
+}
+
+impl Fail for SyncPanicError {}

lib/turnstile/src/job.rs

@@ -0,0 +1,55 @@
+use failure::Fallible;
+use std::time::Duration;
+
+pub trait Job {
+ /// Returns a textual identifier for this job.
+ fn kind() -> &'static str;
+
+ /// Returns true, if this job is due to execute.
+ fn should_run(&self) -> bool;
+
+ /// Returns the execution contract of this job.
+ fn execution_contract() -> ExecutionContract;
+}
+
+#[derive(Debug, Clone, Copy)]
+pub enum Backoff {
+ ConstantWait(Duration),
+ Exponential { base: Duration },
+}
+
+#[derive(Debug, Clone, Copy)]
+pub enum RetryBehavior {
+ Backoff(Backoff),
+ Immediate,
+}
+
+#[derive(Debug, Clone, Copy)]
+pub enum PanicBehavior {
+ Fail,
+ Retry(RetryBehavior),
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct ExecutionContract {
+ pub timeout: Option<Duration>,
+ pub retry_behavior: RetryBehavior,
+ pub autoretry: Option<RetryBehavior>,
+ pub panic: PanicBehavior,
+}
+
+impl ExecutionContract {
+ pub const fn new() -> Self {
+ Self {
+ panic: PanicBehavior::Fail,
+ retry_behavior: RetryBehavior::Immediate,
+ autoretry: None,
+ timeout: None,
+ }
+ }
+}
+
+pub trait Perform {
+ /// Runs this job's action.
+ fn perform(&self) -> Fallible<()>;
+}

lib/turnstile/src/lib.rs

@@ -0,0 +1,7 @@
+mod error;
+pub mod job;
+pub mod worker;
+
+pub use crate::error::Error;
+pub use crate::job::{ExecutionContract, Job, PanicBehavior, Perform};
+pub use crate::worker::Worker;

lib/turnstile/src/worker.rs

@@ -0,0 +1,91 @@
+use serde::de::Deserialize;
+use serde_json::{self, Value};
+use std::collections::HashMap;
+use std::panic::{self, RefUnwindSafe};
+use std::sync::Arc;
+use threadpool::{Builder as ThreadPoolBuilder, ThreadPool};
+
+use crate::error::{Error, SyncPanicError};
+use crate::job::{ExecutionContract, Job, Perform};
+
+type HandlerFn = Box<(Fn(Value) -> Result<(), Error> + Send + Sync + 'static)>;
+type ShouldRunFn = Box<(Fn(Value) -> Result<bool, Error> + Send + Sync + 'static)>;
+
+pub struct Worker {
+ handlers: HashMap<&'static str, Arc<HandlerFn>>,
+ run_checks: HashMap<&'static str, Arc<ShouldRunFn>>,
+ execution_contracts: HashMap<&'static str, ExecutionContract>,
+ thread_pool: ThreadPool,
+}
+
+impl Worker {
+ pub fn new() -> Worker {
+ Worker {
+ handlers: HashMap::new(),
+ run_checks: HashMap::new(),
+ execution_contracts: HashMap::new(),
+ thread_pool: ThreadPoolBuilder::new()
+ .thread_name("worker_thread".to_string())
+ .build(),
+ }
+ }
+
+ pub fn register_job<J>(&mut self)
+ where
+ for<'de> J: Job + Perform + Deserialize<'de> + RefUnwindSafe,
+ {
+ self.handlers.insert(
+ J::kind(),
+ Arc::new(Box::new(|value| {
+ let job: J =
+ serde_json::from_value(value).map_err(|e| Error::DeserializeError(e.into()))?;
+
+ panic::catch_unwind(|| Perform::perform(&job))
+ .map_err(|panic| Error::JobPanicked(SyncPanicError::new(panic)))?
+ .map_err(Error::JobInnerError)?;
+
+ Ok(())
+ })),
+ );
+
+ self.run_checks.insert(
+ J::kind(),
+ Arc::new(Box::new(|value| {
+ let job: J =
+ serde_json::from_value(value).map_err(|e| Error::DeserializeError(e.into()))?;
+
+ Ok(job.should_run())
+ })),
+ );
+
+ self.execution_contracts
+ .insert(J::kind(), J::execution_contract());
+ }
+
+ pub fn job_tick(
+ &mut self,
+ kind: &str,
+ data: Value,
+ on_final: impl Fn(Result<(), Error>, ExecutionContract) + Send + 'static,
+ ) -> Result<(), Error> {
+ let handler = self.handlers.get(kind).ok_or(Error::InvalidKind)?.clone();
+ let execution_contract = self
+ .execution_contracts
+ .get(kind)
+ .ok_or(Error::InvalidKind)?
+ .clone();
+ self.thread_pool.execute(move || {
+ let result = handler(data);
+
+ on_final(result, execution_contract);
+ });
+
+ Ok(())
+ }
+
+ pub fn should_run(&mut self, kind: &str, data: Value) -> Result<bool, Error> {
+ let run_check = self.run_checks.get(kind).ok_or(Error::InvalidKind)?.clone();
+
+ run_check(data)
+ }
+}

migrations/2018-07-31-044227_jobs_table/down.sql

@@ -0,0 +1,2 @@
+DROP TABLE jobs;
+DROP TYPE job_status;

migrations/2018-07-31-044227_jobs_table/up.sql

@@ -0,0 +1,12 @@
+CREATE TYPE job_status AS ENUM ('waiting', 'running', 'dead');
+
+CREATE TABLE jobs (
+ id BIGINT PRIMARY KEY,
+ created_at TIMESTAMP WITH TIME ZONE NOT NULL,
+
+ status job_status NOT NULL,
+
+ queue VARCHAR NOT NULL,
+ kind VARCHAR NOT NULL,
+ data JSONB NOT NULL
+)

migrations/2019-05-14-225400_job_attempt_tracking/down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE jobs
+ DROP COLUMN last_attempt;

migrations/2019-05-14-225400_job_attempt_tracking/up.sql

@@ -0,0 +1,2 @@
+ALTER TABLE jobs
+ ADD COLUMN last_attempt TIMESTAMP WITH TIME ZONE NULL;