Background Jobs

A server often has work to do outside of any request: clean up expired sessions every 15 minutes, send pending notifications, refresh a cache, consume a queue. Sparkle provides two classes for running such work on a background thread:

TJobWorker (unit Sparkle.Host.Jobs) runs a job periodically and on request. This is what most background work needs.
THostedWorker (unit Sparkle.Host.Contracts) runs a loop you write yourself, for work that does not fit a periodic job.

Both run inside a Sparkle host, next to the servers and with the same lifetime, and both also work in any application without a host.

Job Workers

TJobWorker runs a job when it starts, then again TJobWorker.Period seconds after each run finishes, and whenever TJobWorker.Trigger is called.

Creating a job

Derive from TJobWorker, override TJobWorker.ProcessJob, and set the properties in the constructor:

uses
  Bcl.Cancellation,
  Sparkle.Host.Jobs;

type
  TSessionCleanupJob = class(TJobWorker)
  protected
    procedure ProcessJob(const AToken: ICancellationToken); override;
  public
    constructor Create; override;
  end;

constructor TSessionCleanupJob.Create;
begin
  inherited;
  Period := 15 * 60; // seconds
end;

procedure TSessionCleanupJob.ProcessJob(const AToken: ICancellationToken);
begin
  DeleteExpiredSessions(AToken);
end;

Alternatively, set up a job on an instance through TJobWorker.OnProcess, without deriving a class:

Cleanup := TJobWorker.Create;
Cleanup.Period := 15 * 60;
Cleanup.OnProcess :=
  procedure(const AToken: ICancellationToken)
  begin
    DeleteExpiredSessions(AToken);
  end;

Running a job

In a hosted server, add the job to the host builder. The host creates it from the class, starts it after the items added before it, and stops it on shutdown:

uses
  Sparkle.Host;

begin
  CreateHost
    .Add(TApiServerModule)
    .Add(TSessionCleanupJob)
    .Run;
end.

An instance is added through its IHostedService interface. The host then owns it: do not free it yourself.

CreateHost
  .Add(TApiServerModule)
  .Add(Cleanup as IHostedService)
  .Run;

Without a host, call Start and Stop directly and free the worker like any other object. Freeing a worker that is still running stops it first:

Cleanup.Start;
// ...
Cleanup.Free;

How runs are scheduled

Runs never overlap: they happen one at a time, on the worker's own thread. The period is counted from the end of a run, so a slow run delays the next one instead of piling up. A new value assigned to TJobWorker.Period applies to the wait in progress.

TJobWorker.Trigger asks for a run as soon as possible, without waiting for it - for example from an endpoint of your API after new data arrived. If a run is in progress, another one follows it; several requests made meanwhile are served by that single run.

With a period of zero, the job runs when the worker starts and then only when triggered.

Errors

An exception escaping a run is logged through Sparkle.Host.Log, passed to TJobWorker.OnError, and the worker goes on with the next run. A failing job never brings the server down.

Job state

TJobWorker.State returns a TJobRunState snapshot - run and failure counts, when the last run started and finished, how long it took, the last error, when the next run is due - that can be read from any thread, for example by a health check endpoint. TJobWorker.OnRunFinished is called after each run with the updated state.

Custom Workers

For work that is not a periodic job - a loop that consumes a queue, or one with a schedule of its own - derive from THostedWorker and override its Execute method:

uses
  System.SysUtils,
  Sparkle.Host.Contracts,
  Sparkle.Host.Log;

type
  TNightlyJobsWorker = class(THostedWorker)
  protected
    procedure Execute(const AStop: IStopToken); override;
  end;

procedure TNightlyJobsWorker.Execute(const AStop: IStopToken);
begin
  repeat
    try
      RunPendingJobs(AStop);
    except
      on E: Exception do
        if not AStop.IsStopRequested then
          LogError('Pending jobs failed: ' + E.Message);
    end;
  until AStop.WaitFor(60000);
end;

The worker gets a thread of its own on Start. Stop signals the IStopToken and waits for Execute to return. Waiting on the token with WaitFor instead of sleeping is what makes the stop immediate: the request releases the wait at once, wherever the loop happens to be.

An exception escaping Execute is logged and, in a host, brings the whole process down with exit code 1, rather than leaving a server that is up with a worker that died. That is why the loop above catches the exceptions of each iteration: one failed batch should not stop the server.

Like job workers, custom workers are added to the host with CreateHost.Add(TNightlyJobsWorker), or used without a host through Start, Stop and Free.

The Stop Token

The token received by a job (ICancellationToken) and by a custom worker (IStopToken, which descends from it) is a cancellation token, canceled when the worker stops. Everything described in the Cancellation chapter of the TMS BIZ Core Library applies to it. Pass it down into long units of work, so that they give up in the middle instead of holding up the stop:

check ICancellationToken.IsCancellationRequested between steps;
call ICancellationToken.ThrowIfCancellationRequested, which raises EOperationCancelled. Raised after the stop was requested, that exception ends the run or the worker cleanly: it is neither logged nor counted as a failure;
combine it with a time limit of its own through TCancellationTokenSource.CreateLinked.

When the work blocks on something other than the token - a queue, a socket, an event of its own - register a callback with ICancellationToken.Register that wakes that wait up:

procedure TMessageWorker.Execute(const AStop: IStopToken);
var
  Registration: IInterface;
  Msg: string;
begin
  // PopItem knows nothing about the token: shutting the queue down wakes it up
  Registration := AStop.Register(
    procedure
    begin
      FQueue.DoShutDown;
    end);

  while FQueue.PopItem(Msg) = wrSignaled do
    ProcessMessage(Msg);
end;

Migrating from TJobRunner

TJobRunner (unit Sparkle.Sys.JobRunner) is deprecated in favor of TJobWorker. It keeps compiling and working exactly as before, but receives no fixes or improvements. Moving to TJobWorker is mostly a matter of renaming:

// Before
FRunner := TJobRunner.Create;
FRunner.Period := 15 * 60;
FRunner.OnProcess :=
  procedure
  begin
    DeleteExpiredSessions;
  end;
FRunner.Start;

// After
FCleanup := TJobWorker.Create;
FCleanup.Period := 15 * 60;
FCleanup.OnProcess :=
  procedure(const AToken: ICancellationToken)
  begin
    DeleteExpiredSessions;
  end;
FCleanup.Start;

TJobRunner	TJobWorker
`Create`, `Start`, `Stop`, `Free`	Same. `TJobWorker` can also be added to a host.
`Period` (seconds)	TJobWorker.Period (seconds). Zero means "run on `Trigger` only".
`OnProcess: TProc`	TJobWorker.OnProcess, which receives the stop token.
Override `ProcessJob`	Override TJobWorker.ProcessJob, which receives the stop token.
`StopRequested`	The token received by the job: `AToken.IsCancellationRequested`.
`Trigger`	TJobWorker.Trigger
`WaitJob`	No equivalent. Use TJobWorker.OnRunFinished or TJobWorker.State to learn when a run finished.
`LoopCount`	`State.RunCount`
`ErrorMessageFormat`	Errors are written to the Sparkle.Host.Log log; handle TJobWorker.OnError to report them elsewhere.
`StopTimeout`	No equivalent: `Stop` waits for the running job.

Differences in behavior worth knowing:

Stop waits for the running job to finish instead of giving up after a timeout. Check the token in long jobs so that they stop promptly.
Job errors are written through Sparkle.Host.Log instead of Bcl.Logging.
A period of zero does not make the job run continuously.

Table of Contents