aimdb-sync: collapse the bridge onto the block_on seam

[lxsaah]

Background

Most of the sync bridge is already unnecessary. The core API is synchronous on both ends — AimDb::produce() / subscribe() are sync (aimdb-core/src/builder.rs:946, :963), Producer::try_produce is non-blocking, and the buffer reader exposes sync try_recv() alongside async recv() (aimdb-core/src/buffer/traits.rs:179, :189).

Operation	Status in core	Bridge needed?
Produce / non-blocking produce	db.produce / try_produce sync	No — call directly
Non-blocking consume	reader.try_recv() sync	No — poll directly
Blocking consume	only reader.recv().await	Yes — the only runtime-specific piece

So this is simplification, not re-architecture: lean on the sync API, isolate one small seam.

The block_on seam

aimdb-executor::RuntimeOps has no block_on/spawn (aimdb-executor/src/ops.rs:51), so the seam is net-new — but it lives in aimdb-sync, not widened into RuntimeOps.

• Only block_on is needed, never spawn: producer and non-blocking paths call the sync API directly; blocking consume blocks the calling thread.
• The std / embassy features are mutually exclusive per build, so the waiter is a concrete #[cfg]-selected type, not a trait object — no object-safety or generic-dispatch problem.
• #[cfg(feature = "std")] → a waiter holding the tokio::runtime::Handle:
  • get() → handle.block_on(reader.recv())
  • get_with_timeout(d) → handle.block_on(async { tokio::time::timeout(d, reader.recv()).await }). The timed variant must go through the runtime so the tokio timer is in context — this is why a naive futures::block_on will not do.

Handle::block_on is called from the user's sync thread, never from inside the runtime, so this is valid.

Task

• Define the block_on seam as a concrete #[cfg(feature = "std")] waiter holding the Handle.
• Rewrite SyncProducer to call db.produce / Producer::try_produce directly — drop the tokio::mpsc/oneshot round-trip, the forwarder task, and the stored runtime handle for produce.
• Rewrite SyncConsumer to hold the Box<dyn BufferReader<T> + Send> directly and use the seam for get() / get_with_timeout(); try_get() → reader.try_recv(). Delete the per-consumer forwarder task and the std::sync::mpsc::sync_channel.
• Keep the runtime thread + handle + detach/Drop lifecycle — it still hosts the executor driving runner.run().
• Re-audit unsafe impl Send/Sync for AimDbHandle (handle.rs:653) against the much smaller surface. If the justification no longer holds, or is no longer needed, say so explicitly in the PR.

What this does not remove

• The DB's async stages (sources, transforms, taps, connectors) still need an executor driving runner.run(). On std that stays the existing thread.
• Blocking consume still needs a real park/wake primitive per runtime. The seam hides it, it does not eliminate it.

Verification

cargo test -p aimdb-sync
make check

The blocking-API characterization tests must pass unchanged — they are written against the public API precisely so they can arbitrate this rewrite. examples/sync-api-demo behaves identically.

Files

• aimdb-sync/src/{handle,producer,consumer}.rs
• Reference: aimdb-core/src/builder.rs:946/:963, aimdb-core/src/buffer/traits.rs:189, aimdb-executor/src/ops.rs:51

[fresheed]

@lxsaah I'd like to take that one if it's not claimed yet

[lxsaah]

Sure, thanks. There might be dependencies to #199. worst case you can still implement both issues in one PR.

[fresheed]

Regarding SyncProducer::set_with_timeout method: if this producer doesn't have any access to runtime anymore, it won't be able to set up any timeout. Also, the underlying produce won't really block, so timeout doesn't seem to make sense.
Should the method be removed?

[lxsaah]

Good catch! Yes, remove it. set_with_timeout only made sense because the old design could block in the mpsc channel. With produce/try_produce non-blocking and no runtime handle on the producer, there's nothing to time out against.

[fresheed]

Another question about the producer_with_capacity method and whether it still make sense in the new setting. Probably also applies to consumer. I'm trying to wrap my head around the current implementation before making all the changes, so I'd appreciate the input.

For producer, the capacity refers to that of the internal channel used to send a value from the thread calling set to the task spawned on runtime. If that capacity is exceeded, the calling thread blocks. Now, I don't quite get the effect of capacity on the set method: if it's small, the caller will block on sending a value into that channel, but if it's large, the caller will still block afterwards on awaiting the result sent back by the runtime task - it processes incoming values one-by-one, and the capacity doesn't affect the processing speed. So it seems like from the user's PoV the effect of capacity on set is not observable (except the memory consumption). What will however notice the difference is try_set method that immediately returns if the internal channel is full.

Now, in the new setting the internal channel is gone, so I guess the capacity parameter can be removed as well. But if I'm wrong with the intuition above and there is actually an observable difference caused by capacity value, should it be somehow supported in the new implementation?

[lxsaah]

Your intuition is right on both counts. Remove the capacity parameter from the producer and from the consumer. Both channels work around core API limits that no longer exist. When aimdb-sync landed, AimDb::produce was pub async fn and BufferReader had exactly one method for receiving, a boxed async recv() — no try_recv, no poll_recv.

The consumer channel was genuinely needed then. With only an async recv(), try_get() could not be written at all, since block_on(reader.recv()) blocks forever on an empty buffer. A background task pumping values into a std::sync::mpsc::sync_channel was the only way to make them pollable from a sync thread and once it existed for try_get, routing get() and get_with_timeout() through it was consistent.

The producer channel is different. produce was async, so the sync thread had to reach the runtime and that took a channel, a forwarder task, and a oneshot. But the body was a snapshot write and a buf.push(val), an async fn with no .await in it. The round trip was built to reach a function that never yielded.

Both limits are gone: #72 added try_recv to BufferReader and #117 made produce sync — today it is db.produce → writer_handle().push(value). Neither PR touched aimdb-sync, so the bridge kept the machinery. That is the debt this issue collects.

Producer

Your analysis is exactly right. send_internal sends the value and then awaits the oneshot result. So set is a full round trip and there is at most one value in flight per call. A single-threaded caller can never queue more than one item no matter what capacity is set to.

You are also right that try_set is the one place capacity is observable. It drops the result receiver and does not wait, so the queue can fill up and return SyncError::SetTimeout. That signal is worth losing: it reports pressure on the bridge's own queue, not on the database. After the rewrite try_set maps to Producer::try_produce, which returns TryProduceError::Full only when the record's own buffer is bounded and non-overwriting — the default WriteHandle::try_push delegates to push and always succeeds. Real backpressure instead of an artifact, so the difference is replaced rather than lost.

Consumer

Here capacity is observable today, in three ways:

1. get_latest() drains the std channel, so how many values it skips depends on capacity.
2. When the sync consumer is slow, the forwarder keeps draining the record buffer into the std channel, so BufferLagged gets absorbed and logged as a warning instead of reaching the caller.
3. The docs sell capacity=1 as a way to approximate SingleLatest semantics.

All three are consequences of the extra queue rather than features. Once SyncConsumer holds the reader directly:

• Capacity lives at record registration as BufferCfg::SpmcRing { capacity }, which is one knob in the right place instead of two knobs in two places.
• The capacity=1 trick is not needed. Register BufferCfg::SingleLatest and get the real semantics.
• Lag reaches the caller instead of being logged and swallowed.

So please remove consumer_with_capacity as well.

Cleanup that comes with it

drop DEFAULT_SYNC_CHANNEL_CAPACITY from the public API rather than deprecating it and delete the "Channel Capacity Configuration" and "Buffer Semantics Limitation" sections in lib.rs (:129-191) — they exist only to document the extra queue and the second one is an apology for a limitation that stops being true.

try_produce only exists on Producer<T>, not on AimDb. And db.producer(key) resolves the key eagerly (builder.rs:967). So if you pre-resolve a Producer<T> inside SyncProducer, which you need for try_set, then RecordKeyNotFound moves from set() to producer construction. That breaks test_error_propagation (integration_test.rs:507), which currently pins the opposite behaviour — the comment at :516 says so explicitly.

I am fine with that change. The doc comment at handle.rs:294 already promises that producer() returns RecordNotFound for an unregistered type, so pre-resolving makes the code match its documented contract. That test has to be rewritten regardless, since it calls producer_with_capacity and asserts on set_with_timeout. It is the expected exception to "the tests must pass unchanged" — everything else should.

Thanks a lot for your detailed analysis!!!

[fresheed]

Yet another question (I swear I write code as well 🥲).
There seems to be a contradiction on behavior of SyncConsumer's clones: the main spec (under Thread Safety) says they should act as independent readers, but the comment in Clone implementation and the definition of SyncConsumer itself imply that they operate on the same subscription.
The examples I see on multiple readers use multiple .subscribe calls, and the docs for Buffer say "Each subscribe() call returns an independent owned reader". So it seems that for SyncConsumers to be independent, they have to be created independently too.
Supporting cloning for SyncConsumer, and overall "All types are Send + Sync, shareable across threads" guarantee of aimdb-sync - should it still hold? Also, should get of SyncConsumer take &mut self?

[lxsaah]

Keep them coming, this is exactly the kind of reading the crate needed.

You found a doc bug. The comment on the Clone impl is the one telling the truth: every clone shares a single Arc<Mutex<mpsc::Receiver<T>>>, so each value goes to exactly one clone, whichever grabs the lock first. That is a work queue, not independent readers. The line in the struct docs saying "Each clone receives data independently" and the ## Thread Safety section above it, are wrong. You are right that independence comes from calling handle.consumer(key) more than once, because that is what maps to one subscribe() per consumer.

So please drop Clone on SyncConsumer entirely.

Nothing is lost. Independent readers were always the separate consumer() call and anyone who really wants the shared work queue can wrap the consumer in an Arc<Mutex<_>> themselves. There is no reason to make that the default meaning of .clone(). Removing it makes the type say what the docs were already claiming.

&mut self

Yes. Reader<T>::recv and try_recv both take &mut self, so get, try_get, get_with_timeout and get_latest all take &mut self too. Keeping &self would mean putting the Mutex back and that mutex is the thing we are removing. With &mut self the compiler gives us the same exclusion for free.

get_latest still works: block on recv() once, then drain with try_recv() until it comes back empty.

The Send + Sync claim

It needs to become a per type claim, because the two halves really are different.

SyncProducer<T> keeps everything it has. Producer<T> holds an Arc<dyn WriteHandle<T>> and WriteHandle is Send + Sync, so the producer is Send + Sync + Clone on its own. Cloning it is cheap and correct and having many threads produce into one record is fine.

SyncConsumer<T> is Send but not Sync. Reader<T> wraps a Box<dyn BufferReader<T> + Send> and there is no Sync bound on that trait object, so the whole type is not Sync. You can move a consumer to another thread, but you cannot share it between threads. That is the honest guarantee and also the right one, since a reader carries its own receive position and sharing that across threads without a lock does not mean anything.

So lib.rs should say that producers are shareable and consumers are sendable.

One trap to watch for

Please also delete the unsafe impl Send and unsafe impl Sync for SyncConsumer pair at the bottom of consumer.rs. Right now they are only redundant, since Arc<Mutex<Receiver<T>>> gives you both anyway. Once you swap the field over to Reader<T> they turn into a real bug: they claim Sync for a type that is not Sync, the compiler accepts it without a word and we end up with undefined behaviour that no test will catch. The SyncProducer pair can go for the same reason, since Producer<T> gets both on its own.