Grdb.swift: Viability of a job queue using ValueObservation

Created on 8 Jan 2021 · 3Comments · Source: groue/GRDB.swift

👋

What did you do?

I'm working on adding a basic job queue backed by sqlite to an ios application. Right now I'm evaluating using ValueObservation for getting the jobs to run. I have something like this in mind:

/// Reaction to a profile, either a "like" or a "dislike"
struct ProfileReaction {
  let reaction: Reaction // .like or .dislike
  let profile: UUID
}

/// Feed job queue watches for new scheduled ProfileReaction jobs and tries to execute them. Once the job is executed successfully, it is removed from the database.
final class FeedJobQueue {
    private let dbQueue: DatabaseQueue
    private var cancellable: DatabaseCancellable?

    init(dbQueue: DatabaseQueue) {
        self.dbQueue = dbQueue
    }

    func start() {
        let observation = ValueObservation.tracking { db in
            return db.execute(sql: "select id, reaction, profile_id from profile_reactions where status != 'running'")
        }

        let queue = DispatchQueue(label: "app.example.feed_job_queue")

        self.cancellable = observation.start(
            in: dbQueue,
            scheduling: .async(onQueue: queue),
            onError: { error in fatalError("profile_reactions couldn't be fetched \(error)") },
            onChange: { (reactions: reactions) in
              try dbQueue.write { db in
                try db.execute(
                  sql: "update profile_reactions set status = 'running' where id in $0",
                  arguments: [reactions.map { $0.id }]
                ) 
              }
              execute(reactions) // on successful execution, the job is removed from profile_reactions, and on fail, it's status changes to `failed`
            }
        )
    }
}

Question

I wonder if the approach above can result in duplicate executions for the same job due to some race condition? And also is it possible to not get a notification about a new job in onChange?

Environment

I don't think it's applicable, but I can fill these out if needed :)

GRDB flavor(s): (GRDB, SQLCipher, Custom SQLite build?)
GRDB version:
Installation method: (CocoaPods, SPM, manual?)
Xcode version:
Swift version:
Platform(s) running GRDB: (iOS, macOS, watchOS?)
macOS version running Xcode:

P.S.

Thank you for your work on GRDB!

support

Source

ruslandoga

All 3 comments

Hello @ruslandoga,

So you observe pending reactions and turn them to running. All right.

The code itself contains several errors: you don't return any fetched value from the ValueObservation, and your update won't compile because IN $0 is not a valid way to feed a request with an array.

But these problems in your sample code (there are maybe others) are not what this issue is about. You'll deal with them in due time.

I wonder if the approach above can result in duplicate executions for the same job due to some race condition?

Good question. The code does not do what you want.

Look at this scenario:

App inserts a pending job A.
-> ValueObservation is kicked, and will notify the pending job [A] (N1)
App inserts a pending job B.
-> ValueObservation is kicked, and will notify the pending jobs [A, B] (N2)
N1 [A] is received: job A is started and set to running.
-> ValueObservation is kicked, and will notify the pending job [B] (N3)
N2 [A, B] is received: job A is started (again), job B is started, both are set to running.
-> ValueObservation is kicked, and will notify the pending job [] (N4)
N3 [B] is received: job B is started (again) and set to running.
N4 [] is received.

Both jobs A and B have been started twice!

To fix this, you can only start jobs that were not already started, by checking their state in the database (inside the transaction that sets them to running - transactions are a great tool to avoid race conditions):

App inserts a pending job A.
-> ValueObservation is kicked, and will notify the pending job [A] (N1)
App inserts a pending job B.
-> ValueObservation is kicked, and will notify the pending jobs [A, B] (N2)
N1 [A] is received: job A is started and set to running.
-> ValueObservation is kicked, and will notify the pending job [B] (N3)
N2 [A, B] is received: job A is not started again, job B is started and set to running.
-> ValueObservation is kicked, and will notify the pending job [] (N4)
N3 [B] is received: job B is not started again.
N4 [] is received.