fix(etl): skip errored state and interrupt flush wait on table-copy shutdown - #881
fix(etl): skip errored state and interrupt flush wait on table-copy shutdown#881abhizer wants to merge 1 commit into
Conversation
|
Hi, thanks for this PR! I will get to it once I find the time. |
|
Thanks! |
…hutdown Currently, during table-copy, if the destination drops the AsyncResult while the pipeline is shutting down, the table ends up in an Errored state. But dropping the AsyncResult during shutdown should be safe: on restart, we should just ensure the retry copies the table from the start. Also make the copy loop stop waiting on the flush result once shutdown fires, instead of blocking on a destination that never completes it. Signed-off-by: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com>
5a9d35f to
6d33c24
Compare
|
Hey @abhizer I think I got this issue last week too. What was the error you got once it happened ? |
|
Coverage is 70.861% — abhizer:fix/table-copy-shutdown-write-race into supabase:main. No base build found for |
| // a stored `Errored` state is never retried across restarts, so the | ||
| // table would otherwise stall on every later run. A dropped sender also | ||
| // counts as shutdown. | ||
| if shutdown_rx.has_changed().unwrap_or(true) { |
There was a problem hiding this comment.
This is not the correct solution for a concurrent system like ETL. Having the shutdown being true at this phase, doesn't guarantee us that the error will be caused by the shutdown procedure.
There was a problem hiding this comment.
That's fair. We'd just like to be able to drop the acknowledgements once we've initiated the shutdown.
There was a problem hiding this comment.
Can you give me more context on your ETL use case? So that I can see how we can better design shutdown in case.
There was a problem hiding this comment.
The reason of why I am saying this is that, the shutdown on the destination is called on purpose after all of ETL is done, so that we can stop producing data on our end and then the destination can perform teardown.
The failure should not happen because a destination is technically unaware of shutdown until ETL won't be interested anymore about the result. And if the result is sent back and the channel is closed a warning will be raised.
if tx.send(result).is_err() {
warn!("could not send async result because receiver was already closed");
}There was a problem hiding this comment.
Our use case is the Postgres CDC input connector for feldera.
We want the connector to snapshot, and then follow the cdc stream, this works.
But we'd also like to ensure fault tolerance and be able to reingest the table in cases of failure. In cases of failure, the connector sees that the pipeline is shutting down, and triggers the etl pipeline to shutdown as well. And in such cases, returns from write_table_rows without acknowledging the AsyncResult, which can lead to the table being in errored state.
Because we've already requested a shutdown here, we want this to be okay.
Related: The Accept / Durable api for table copy, currently on Accept can currently allow the table to transition to Ready, causing etl to skip reingesting on restart. I do not think that Accept should have this durable side effect.
There was a problem hiding this comment.
Ah yeah, I get what you mean. The thing here is that we need to clarify the semantics. I feel like from ETL's side, if a channel is closed without a response, it's a problem. The destination should take care of sending back a response. Maybe we could classify a response as "gracefully stop". But from my idea, I would like the system to be like, if there is shutdown after the write_table_rows method, we immediately return. Would that work?
There was a problem hiding this comment.
Sorry, I don't think I quite understand the immediately return model. Do you mean something like a biased tokio select, on shutdown.changed() and the write_table_rows method?
But yes, the gracefully stop idea, for shutdown during write_table_rows, returning an ErrorKind::DestinationShutdown should work well for us, and in other similar use cases. ETL could then treat it as a graceful cancellation, instead of an error.
There was a problem hiding this comment.
One question I would have is. I assume your write_table_rows fails because you have an out-of-bound shutdown signal receiver in your destination which makes it stop and not return a result?
The reason I am asking is that ETL is designed in a way where the shutdown procedure of a destination should be made in the shutdown() method, so that teardown is properly controlled.
There was a problem hiding this comment.
One question I would have is. I assume your write_table_rows fails because you have an out-of-bound shutdown signal receiver in your destination which makes it stop and not return a result?
Yes. As the connector is part of the pipeline, the shutdown procedure cannot quite be contained in the shutdown() method. We need to shutdown the etl pipeline safely, when shutting down the outer pipeline.
There was a problem hiding this comment.
This would make the implementation quite a bit uglier, is this a problem also for the apply loop?
| .write_table_rows(&replicated_table_schema, table_rows, flush_result) | ||
| .await?; | ||
| let write_status = pending_flush_result.await.into_result()?; | ||
| let write_status = tokio::select! { |
There was a problem hiding this comment.
This is fair but I think it would be better if we make an abstraction which behind the scenes waits for shutdown on a result.
There was a problem hiding this comment.
I am working now on a PR to do that.
@bnjjj I don't exactly remember, but it was something to do with the table being in errored state, when we've tried to shutdown the etl pipeline, and then dropped the async result. |
This is interesting. The only way in which it could have happened is if the |
|
Yes probably something in the ducklake destination I'll check |
4 participants
Currently, during table-copy, if the destination drops the AsyncResult while the pipeline is shutting down, the table ends up in an Errored state. But dropping the AsyncResult during shutdown should be safe: on restart, we should just ensure the retry copies the table from the start.
Also make the copy loop stop waiting on the flush result once shutdown fires, instead of blocking on a destination that never completes it.
What kind of change does this PR introduce?
Bug fix
What is the current behavior?
During table-copy, if the destination drops the AsyncResult while the
pipeline is shutting down, the table ends up in an Errored state. The
copy loop also waits on the flush result unconditionally, so it can
hang if the destination never completes it.
What is the new behavior?
Dropping the AsyncResult during shutdown no longer persists an Errored
state; the table retries the copy from the start on restart. The copy
loop also stops waiting on the flush result once shutdown fires.