RFC - Background Jobs Manager

[Datron]

Superposition has invisible background jobs that happen in the system like webhook calls, priority recomputations, etc. which a user cannot see nor inspect, unless they have access to the infrastructure superposition is run on.

This proposal aims to define a structure for the background job manager (BJM), the interaction of BJM with Kronos, the types of background jobs that should be supported and where current invisible jobs get classified into.

List of all background jobs

Currently active

• webhooks for contexts and experiments
• priority recompute when dimensions change priority
• reduce (remove redundant overrides)

Jobs Enabled By BJM

• update all workspace schemas in superposition for a new version
• import data into a workspace
• export data to a workspace
• run integrity checks on contexts & overrides when a dimension or default config is changed
• update contexts when dimension values change
• update overrides when default config values are changed
• bulk update/delete

Types of background jobs

Background jobs are classified by whether they will need to modify data in tables in workspace schema.

Modification Jobs

These are jobs that will be doing writes to the workspace schema over a period of time and will be impacted if any parallel changes occur while this happens. They will need to lock the workspace and for safety, set the config version of the workspace to the last working version (latest before changes). This guarantees no issues for clients or services resolving configs using superposition.

From the jobs list, the ones that fall into this are:

• import data into a workspace
• run integrity checks on contexts & overrides when a dimension or default config is changed
• update contexts when dimension values change
• update overrides when default config values are changed
• bulk update/delete
• priority recompute when dimensions change priority
• reduce (remove redundant overrides)

Processing jobs

These are jobs that read data and make API calls, but make no writes to the database with the exception of the bjm table.

From the jobs list, the ones that fall into this are:

• webhooks
• export
• dry run of integrity checks on contexts & overrides when a dimension or default config is changed
• dry run of import data into a workspace
• dry run of bulk update/delete

Background Job Manager

The BJM table will exist in the global superposition schema. The table schema is below:

CREATE TYPE public.background_job_type AS ENUM (
    'MODIFICATION',
    'PROCESSING'
);

CREATE TYPE public.background_job_status AS ENUM (
    'CREATED',
    'SCHEDULED',
    'INPROGRESS',
    'FAILED',
    'COMPLETED'
);
    
CREATE TABLE IF NOT EXISTS superposition.job_manager (
  job_id BIGINT PRIMARY KEY,
  kronos_job_id TEXT NOT NULL,
  job_type BACKGROUND_JOB_TYPE NOT NULL,
  job_status BACKGROUND_JOB_STATUS NOT NULL,
  job_name VARCHAR(15) NOT NULL,
  progress INT NOT NULL DEFAULT 0,
  workspace_schema TEXT,
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  created_due_to TEXT NOT NULL
 );

The workspace schema is NULLABLE because there are jobs that may exist to make workspace management easier. VIEWS of this table will be created in every workspace with the workspace_schema filter, so that workspace users can view the status of jobs without breaking data isolation. This may not be needed if we want to skip it, but is a nice to have.

How it works

Job scheduling and retries are managed by Kronos, and to show these attempts/setup/details to the user Superposition will reach out to kronos APIs to get the data using the kronos_job_id. The BJM only tracks the status of the job triggered in Superposition and its status and outcome. When a background job is scheduled in kronos, a corresponding job tracker will be added to BJM in superposition with the status CREATED. If kronos responds with a 201 accepting the job, then the status is changed to SCHEDULED.

It is recommended that background jobs are atomic and use transactions, but the design plans for cases where code must commit changes to the DB. To handle such cases or failures, all modification jobs when they start will set a config version for the workspace. When the job is triggered, they update the status to INPROGRESS and based on the outcome will set the status to either FAILED or COMPLETED. The job can reschedule itself in case it fails, and the cycle repeats.

[sauraww]

Would we set up some retention policy ?
I mean do we keep the completed jobs indefinitely ?

[Datron]

We can, or that can be something the DBA decides?

[sauraww]

Can we add an enum for job name as well ?

[Datron]

A DB enum? we can think about it - a rust enum is enough me thinks

[knutties]

@Datron - could you add some examples of job_name?

I am wondering what field will hold the information that it is a webhook/bulk/priority_recompute etc. ? Is that the name ?

[sauraww]

I think that is what it should be :

1. IMPORT
2. EXPORT
3. WEBHOOK
4. PRIORITY_RECOMPUTE
5. REDUCE
6. BULK_UPDATE
7. INTEGRITY_CHECK

[knutties]

@sauraww - that feels more like type than name ?

[sauraww]

then , why do we need this :

CREATE TYPE public.background_job_type AS ENUM (
    'MODIFICATION',
    'PROCESSING'
);

I think job is best described by the names itself.
we can have them in the types itself.

[sauraww]

Job type and job name will give us full clarity. Also, having job name as enum would ensure that we don't have arbitrary strings.

[Datron]

we can think about it - a rust enum is enough me thinks

[sauraww]

created_due_to looks a bit odd, Can we do something like:
trigger_reason and triggered_type
triggered_type can be an enum :
trigger_type -- USER | SYSTEM | SCHEDULED | EVENT

[Datron]

Sure

[sauraww]

Are we planning to store job specific metadata also ?

[sauraww]

For import let's say it can be

• import mode
• total rows
• maybe uploaded file ID

for bulk update:

• operation types
• total records

[Datron]

Audit log would record these operations, would it be valuable to record these again?

[sauraww]

How would that relate to the job ? I mean how would they correlate that this audit table is created due to this job ?

[Datron]

The database changes that the job does will be in audit log. If we want any other metadata that cannot be found anywhere, we can always add the column later. This will be based on if we need it when writing import/bulk update job code

[Datron]

I mean how would they correlate that this audit table is created due to this job ?

Description, change reason, user, etc

[knutties]

We can remove the job_ prefix I guess.

CREATE TABLE IF NOT EXISTS superposition.job_manager (
  id BIGINT PRIMARY KEY,
  kronos_job_id TEXT NOT NULL,
  type BACKGROUND_JOB_TYPE NOT NULL,
  status BACKGROUND_JOB_STATUS NOT NULL,
  name VARCHAR(15) NOT NULL,
  progress INT NOT NULL DEFAULT 0,
  workspace_schema TEXT,
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  created_due_to TEXT NOT NULL
 );

[knutties]

created_due_to will be filled with change_reason ?

[sauraww]

That's what I also though first , but wouldn't be a good fit for these kinds of jobs.
1> export
2> webhook delivery
3> dry run

[Datron]

Yeah we have change_reason in a few tables where we just put dummy strings. Its not useful in these contexts

[Datron]

After discussions, some changes:

• Jobs are no longer divided into PROCESSING and MODIFICATION, they will instead be categorized by the feature - WEBHOOKS, REDUCE, etc for job_type
• created_due_to will be replaced by job name
• add a description so code can write some details about the job
• add a boolean column on whether the job will lock the workspace or not - workspace_lock
• workspace_schema will not be null for global schemas, instead we will use a enum GLOBAL
• limit progress between 0 to 100
• introduce logs in the table so that job execution details can be shown in frontend

CREATE TYPE public.background_job_type AS ENUM (
    'WEBHOOK',
    'PRIORITY_RECOMPUTE',
    'REDUCE'
);

CREATE TYPE public.background_job_status AS ENUM (
    'CREATED',
    'SCHEDULED',
    'INPROGRESS',
    'FAILED',
    'COMPLETED'
);
    
CREATE TABLE IF NOT EXISTS superposition.job_manager (
  id BIGINT PRIMARY KEY,
  kronos_job_id TEXT NOT NULL,
  description TEXT NOT NULL,
  type BACKGROUND_JOB_TYPE NOT NULL,
  status BACKGROUND_JOB_STATUS NOT NULL,
  name TEXT NOT NULL,
  progress INT NOT NULL DEFAULT 0 CHECK (progress >= 0 AND progress <= 100),
  workspace_schema TEXT NOT NULL,
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  logs TEXT NOT NULL
 );

Sequence Diagram

functions:

execute_job: execute job will handle all job API calls from kronos. Today for a webhook, we have an endpoint to handle them - dispatch_handler that takes the request:

#[derive(Deserialize)]
struct DispatchWebhookRequest {
    webhook_name: String,
    data: serde_json::Value,
}

This will be replaced by job_handler that takes the generic JobRequest

enum JobRequest {
 Webhook(DispatchWebhookRequest { webhook_name: String, data: serde_json::Value }),
 ...,
 ...
}

(unsure but it'll be something like this in rust, will work it out when I write it)
The code to handle this will be defined as:

fn dispatch_webhook(req: &DispatchWebhookRequest) -> response 

Finally, execute_job will look like:

execute_job(req: &JobRequest) -> Response {
match req {
 Webhook(r) => dispatch_webhook(r),
 ....
 ....
 }
}

This is subject to change, I'll bring up a prototype soon