Skip to main content

Design

Where ruststream-sqlx sits: between the RustStream runtime and the service's sqlx pool, with SQL built once per subscription and the outbox beside any broker.

The service, the RustStream runtime, this crate, the service's sqlx pool, and the database, top to bottomYour serviceRustStream coreruststream-sqlxsqlx#[subscriber]handlersRustStream::new(..).with_broker(..)Runtime: dispatch, workers, settlement, publishInboxSubscriberthe claim loopInboxDeliverysettles its rowRepository, Routedinsert rowsInboxQueue<Row>descriptorSqlxBrokerconfigurationDialectSQL per databaseOutboxlayers, any brokerPool<DB>built at connectPostgreSQL, MySQL, MariaDBor SQLiteUnder the testing feature, TestApp connects in process:the same pool and database, on the test's clock
The diagram as text
  1. Your service: #[subscriber] handlers; RustStream::new(..) .with_broker(..)
  2. RustStream core: Runtime: dispatch, workers, settlement, publish
  3. ruststream-sqlx: InboxSubscriber the claim loop; InboxDelivery settles its row; Repository, Routed insert rows; InboxQueue<Row> descriptor; SqlxBroker configuration; Dialect SQL per database; Outbox layers, any broker
  4. sqlx: Pool<DB> built at connect
  5. #[subscriber] handlers → Runtime: dispatch, workers, settlement, publish: mounted
  6. Runtime: dispatch, workers, settlement, publish → InboxSubscriber the claim loop: polls
  7. InboxSubscriber the claim loop → Pool<DB> built at connect
  8. Repository, Routed insert rows → Pool<DB> built at connect
  9. Pool<DB> built at connect → PostgreSQL, MySQL, MariaDB or SQLite: the driver
  10. Under the testing feature, TestApp connects in process: the same pool and database, on the test's clock

The crate implements the core's broker traits over sqlx. The runtime above it is the same for every broker. Everything SQL-specific is in this crate: the InboxQueue descriptor, the claim loop, the three claim forms, what an outcome writes, and the SQL of each database. The crate subscribes, by descriptor and by name, takes batches, hands a row-mode handler the row itself, hands the partition key to keyed dispatch, publishes into tables, and describes its server in the generated document. It has no request-reply and no transactional publisher: transactional mode is a step of a subscription, where a handler writes through its delivery's transaction.

The pool's settings are the service's: its size, its options and its migrations. SqlxBroker::from_url records them without I/O, so the app is built before the runtime starts; the broker builds the pool when it connects and closes it at shutdown. A service that shares one pool with code of its own builds it in its main, inside the runtime, and hands it to SqlxBroker::new, which never closes it.

The SQL​

A dialect builds every statement a subscription runs: PostgreSQL, MySQL and MariaDB, and SQLite are built in, an AnyPool takes the one of the database it reaches when the broker connects, and a service may bring a dialect of its own for another database. A subscription builds its statements once, when it opens, from the table's description, a const value the derive or the hand-written InboxSpec produces. So a message costs the same with the derive, the manual description or a dialect of the service's own. The startup then prepares each statement on the server, so a missing table or column stops the service before the first message.

Publishing​

A retry stays in its row. The descriptor declares that the database moves a delivery itself, so a retry_after is an UPDATE of the row's retry_after column, and no copy is ever published: Settlement draws it. A publish writes a row through the row type's own Publish impl. A Repository names its table at compile time. Routed, the connected broker's default publisher and so the one replies take, looks the message's name up in the routes the broker was built with. Both take the encoded bytes as a lent slice, so a reply's buffer is reused from one message to the next.

What remains per message is stated by the crate and measured on Benchmarks:

  • Routed makes one hash lookup and one dynamic call into the row's Publish. The call's future stays in a 1024-byte slot in place, and is boxed only when it is larger.
  • A publish reads the broker's closed flag, one atomic load, and wakes the subscriptions of its table in the process, one atomic operation each.
  • A handler that reads Tx or Ctx<keys::Pool> pays one reference-count increment per delivery.
  • A by-name subscription over a row whose events are the service's own pays one boxed delivery and one boxed settlement future per message.

In-process mode​

Under the testing feature, TestApp::start connects the broker in process. Nothing below the pool is replaced: the in-process broker reaches the database the test names, runs the service's SQL, its schema and its claim forms, and fails where the database fails. SQLite in memory serves a test with no server. What changes is time. The broker reads "now" from the tokio clock, anchored at the wall clock when it connected, so a test on a paused clock moves the queue's time with tb.advance(..): a retry_after row comes due when the test advances past it. Every database call runs off that clock, so a paused clock does not jump to its next timer while a reply from the database is on its way. The test clock stands in for SystemClock alone: a table on DatabaseClock reads the database's clock, and a clock of the service's own stays in force, as in production.

The harness waits on the database's work through the broker's books. It counts a message when it lands in a queue a subscription of the test reads, and again when a retry returns its row, so tb.settle() returns once every row that reached a queue was delivered and settled.

How the pieces start and stop is on Lifecycle, and the way of one row on Path.