implement pause and resume data ingest #134
Conversation
pkg/workers/workers.go
Outdated
| } else { | ||
| message, err := w.messageToStruct(item) | ||
| select { | ||
| case <-ctx.Done(): |
There was a problem hiding this comment.
I actually intentionally don't use a channel to dequeue messages, and here is why. (tell me if there is a logic error.) Suppose this sequence of events:
- queue.Dequeue() fetches a message from SQS
- We attempt to send that message to the channel, but it blocks because the worker is busy
- We ctrl-c the program
In this case, I think we'd lose a message. ctx.Done() will trigger, this worker will return and the program will terminate. Meanwhile we've already fetched a message from SQS, but it hasn't been consumed by anyone.
(No need to think about this now, just something to talk through!)
There was a problem hiding this comment.
yes, this is a failure scenario. this is actually something I noticed in the code. The message should only be deleted from the queue if it is handled successfully https://chat.openai.com/share/1b1b2c58-6ee5-4b3f-b0f3-67260eb71780
The queue delete needs to be moved out of this function and is called after a successful processing https://github.com/scratchdata/scratchdata/blob/main/pkg/storage/queue/sqs/sqs.go#L75
There was a problem hiding this comment.
That seems like a lot of work just so we can a channel, doesn't it? To me, the question is whether we want consumers from the queue to have to contend with a Delete() operation in addition to Dequeue?
Other things to think about: What timeout should we set before SQS automatically re-queues an unacknowledged message? If we end up using Redis as a queue in the future (ie, BLPOP) then consuming+deleting are the same action.
There was a problem hiding this comment.
even in the case of redis, you would want to use BRPOPLPUSH to guarantee processing https://chat.openai.com/share/bcac4373-b4dc-4672-9f7e-c81ed984b209. I think the only way to implement this is to have an initial "acknowledgement" the item has been observed, and then an explicit "I'm done" by removing it from the queue, for example. And then in the failure case, it needs to go somewhere to be "re-acknowledged". Optimizing for durableness can be a rabbit hole, probably best to sketch out what the requirements are and pair that with some tech decisions that match those.
|
Quick thought: I wonder if we could use dead-letter queues to implement a pause functionality by toggling where new messages are sent. |
|
If you reject a message every time it belongs to a paused table, it will go to the back of the queue. A DLQ would be helpful if we want to keep track of paused queues that are putting pressure on the queue. |
I worry about chattiness+starvation here: re-queuing items over and over again, and consuming workers with the busywork of resending paused items. I'm wondering if SQS is the right queue to use after all. |
|
This is another YC company (also go OSS): https://hatchet.run/ they use rabbitmq currently for their queue. |
|
from our conversation, we concluded that existing throughput can be handled by an SQL queue until we "feel the pain". When we hit that point, redis could be an option to explore. |
Closes #131
Currently, a new queue is made for every table that has items to process. In the case of sqs, this will not work properly unless a new queue is made for every table. An alternative approach to mitigate overhead of sqs queue deployments, would be to attach the name of the table to the queued item, deqeue, check if it belongs to a paused table, if it doesn't then process, otherwise put it back on the queue. Further clarification would be needed to address potential concerns of this approach.
Example flow: