How do I write a background service that restarts when notified by API methods?

Viewed 50

I have a background service that does some processing of records in a DB. If there are no records left to process, the service should be idle. An HTTP POST method in the application to endpoint X may allow for a new record to be inserted, necessitating that the background service starts running again.

Importantly, records must be processed sequentially, according to some order. These jobs cannot be parallelised, as they affect the state of an external system.

It's possible for a job to be paused due to some condition not being met. In this case, other jobs should not start, as they may depend on the first one being completed.

Processing of each record may take some time, so if two records are inserted in quick succession the background service should ignore the notification output by the second insertion, as it will likely still be processing the first. I do not want to busy-wait, e.g.

while (true)
{
    if (db.HasRecordsToProcess())
    {
        _recordProcessor.ProcessRecords();
    }
}

nor do I want to Thread.Sleep, i.e.

while (true)
{
    if (db.HasRecordsToProcess())
    {
        _recordProcessor.ProcessRecords();
    } else
    {
        Thread.Sleep(500);
    }
}

Instead, I want the service to do nothing until the API method tells it to wake up. Below is some of the code I've written.

    public class MyBackgroundService : BackgroundService
    {
        private readonly IRecordProcessor _recordProcessor;
        private readonly AutoResetEvent _isRunning = new AutoResetEvent(true);
        private readonly ILogger _logger;

        //... constructor...

        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            await Task.Yield(); // Required for the method to be executed asynchronously on startup.            
            while (!stoppingToken.IsCancellationRequested)
            {
                _isRunning.WaitOne();
                _recordProcessor.ProcessRecords(); // do some long-running work
            }
        }

        /// <summary>
        /// Notifies the background service to resume processing records.
        /// </summary>
        public void Resume()
        {
            _isRunning.Set();
        }
    }
    public interface IRecordProcessor
    {
        void ProcessRecords();
    }

The implementation of IRecordProcessor looks something like this:

        public void ProcessRecords()
        {
            var isRunning = true;
            while (isRunning)
            {
                // this is basically the db.HasRecordsToProcess bit from the pseudocode in the 
                // first two examples of my question, showing busy-waiting / Thread.Sleeping
                var queueHead = db.GetRecords.OrderBy(m => m.StartTime).FirstOrDefault();

                isRunning = queueHead != null;
                if (!isRunning) continue;
                heavyweightProcessor.Process(queueHead); // this might take a few minutes
            }
        }

The implementation thus handles a case where a record is being processed, and while that's happening another record is inserted. When it loops again, it will find the other record and just continue processing.

An API method where the background service can be told to restart:

[HttpPost("record", Name = "CreateRecordToProcess")]
public ActionResult<CreatedRecord> CreateRecordToProcess()
{
    var record = _someDbService.CreateSomeResource();
    _backgroundService.Resume();
    return CreatedAtRoute("GetRecordToProcess", new { id = record.Id}, (Record) record);
}

Is there a more idiomatic way to define this behaviour in .NET?

As mentioned, there are a couple of cases where I want well-defined behaviour:

  1. If the background service is processing a record, and a second record is inserted, then the service should carry on as if it received no notification. By extension, if the service is processing a record and N new records are inserted for it to process, it

  2. If the background service is not processing a record, it should wait (without Thread.Sleep or busy-waiting) until a relevant resource is saved, at which point it should be notified -- by the API method that saved the resource -- to start querying the DB again to get records to process.

How do I test this?

Test at the moment:

var mockProcessor = new MockRecordProcessor(_testOutputHelper);
_backgroundService = new MyBackgroundService(...otherDependencies, mockProcessor);

var token = new CancellationToken();
await _backgroundService.StartAsync(token);
Thread.Sleep(10_000);
_backgroundService.Resume();
Thread.Sleep(10_000);
Assert.Equal(2, mockProcessor.Count);

I would obviously prefer not to do the Thread.Sleep in this test, but I can't work out how to specify a condition in the test like "wait for initial processing to finish before explicitly notifying the service to resume with _backgroundService.Resume".

0 Answers
Related