refactor(sinktools): LazySinkSource cleanup waker handling, use StreamExt::split, improve unit tests, fix #2336

[MingweiSamuel]

Fix #2336

Adds a couple more simpler tests

[cloudflare-workers-and-pages]

Deploying hydro with    Cloudflare Pages

Latest commit:	68920b6
Status:	✅  Deploy successful!
Preview URL:	https://3c5a5e5d.hydroflow.pages.dev
Branch Preview URL:	https://lazy-1.hydroflow.pages.dev

View logs

[Copilot]

Pull request overview

Refactors sinktools’ LazySinkSource to simplify/optimize wakeup handling (addressing #2336) and modernize splitting by relying on StreamExt::split, alongside strengthening initialization-driving test coverage.

Changes:

• Replace the old multi-waker approach with a 2-slot DualWaker based on futures_util::task::AtomicWaker.
• Refactor LazySinkSource to hold state directly (removing Rc<RefCell<...>>) and remove the custom split-half types in favor of StreamExt::split.
• Add/adjust unit tests for initialization behavior; update dev-deps to enable tokio::time.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 8 comments.

File	Description
sinktools/src/lazy_sink_source.rs	Refactors internal state + wakeup handling; switches to StreamExt::split; expands unit tests.
sinktools/Cargo.toml	Enables tokio’s time feature for dev-dependencies to support new tests.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated 4 comments.

Comments suppressed due to low confidence (1)

sinktools/src/lazy_sink_source.rs:80

• Future is referenced in the trait bounds (Fut: Future<...>) but isn't imported in this module, which will fail to compile unless it happens to be in scope. Add an explicit use core::future::Future; (or std::future::Future) near the top of the file.

impl<Fut, St, Si, Item, Error> Sink<Item> for LazySinkSource<Fut, St, Si, Item, Error>
where
    Self: Unpin,
    Fut: Future<Output = Result<(St, Si), Error>>,
    St: Stream,
    Si: Sink<Item>,
    Error: From<Si::Error>,
{

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[MingweiSamuel]

Docker tests pass locally