fix(replication): replace custom AbortToken with tokio::CancellationToken

[meskill]

Replace custom AbortSignal with CancellationToken to make the cancellation explicit and not relied on futures drop (that holds rx part of channel). Pass cancellation token for graceful cancellation. Early cancellation on errors

[codecov]

Codecov Report

❌ Patch coverage is 43.39623% with 30 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
...end/replication/logical/publisher/parallel_sync.rs	0.00%	13 Missing ⚠️
...nd/replication/logical/publisher/publisher_impl.rs	76.66%	7 Missing ⚠️
...og/src/backend/replication/logical/orchestrator.rs	0.00%	5 Missing ⚠️
pgdog/src/api/copy_data.rs	0.00%	3 Missing ⚠️
...src/backend/replication/logical/publisher/table.rs	0.00%	2 Missing ⚠️

📢 Thoughts on this report? Let us know!

[sgrif]

Suggested change

	// If every handle succeed the guard token will just cancel already finished work
	// If every handle succeeds the guard token will just cancel already finished work

[sgrif]

I think we should require the caller to call .child_token() if they don't want to be cancelled.

Suggested change

let cancel = cancel.child_token();

[meskill]

I think it's better to leave it here, in the ParallelSyncManager. In the run method we start multiple ParallelSync instances to copy tables in parallel. So, basically ParallelSyncManager manager the parallel work here and the cancel token is here to help manage it: we create a child_token with guard to not affect parent or sibling work, but still be affected if the parent is cancelled. The guard is needed to cancel all the other running syncs in case any of them fails: in that case we cancel this execution subtree and propagate the error to parent where the parent will react on error and finish with the error eventually and not cancellation.

[sgrif]

Suggested change

	let _guard = cancel.clone().drop_guard();
	let _guard = cancel.drop_guard_ref();

[sgrif]

Suggested change

	let cancel = cancel.clone();
	syncs.push(async move {
	let manager = ParallelSyncManager::new(tables, replicas, dest)?;
	let tables = manager.run().await?;
	let tables = manager.run(cancel).await?;
	syncs.push(async move {
	let manager = ParallelSyncManager::new(tables, replicas, dest)?;
	let tables = manager.run(cancel.child_token()).await?;

[sgrif]

This is the first place we actually do anything with a cancellation token other than pass it through. Should we remove everything up the chain from here and the CancellationToken::new()s, and create it at this point?

[meskill]

The idea here, that the initial CancellationToken comes from the

pgdog/pgdog/src/api/async_task.rs

Lines 358 to 361 in 02c7a32

	cancellation_token: match parent_task {
	Some(parent) => parent.cancellation_token.child_token(),
	None => CancellationToken::new(),
	},

so the whole task becomes manageable from there. We create the token in the root and pass it down to any child task. That's why it passed here and just cloned or child_token is created.