From 80cf2930a0d648ec8f6fa44cea11c02b418961f5 Mon Sep 17 00:00:00 2001 From: Haokang Den Date: Tue, 27 Oct 2026 18:51:24 +0900 Subject: [PATCH] feat(turso): support bounded long polling Allow official sync clients to wait for updates without accepting unbounded client timeouts. Keep the published SlateDB binding and bound disconnected requests without adding notification state. --- compose.yaml | 1 + docs/testing.md | 1 + docs/turso-compatibility.md | 32 ++++++++- duct.edn | 6 +- src/linear/adapter/slatedb/store.clj | 16 ++++- src/linear/handler/turso/protobuf.clj | 2 + src/linear/handler/turso/pull.clj | 2 +- src/linear/usecase/core.clj | 10 +++ src/linear/usecase/database.clj | 21 +++++- src/linear/usecase/database/revisions.clj | 13 ++++ test/e2e/turso-sync.mjs | 69 ++++++++++++++++++ test/linear/e2e.clj | 3 +- test/linear/handler/turso/long_poll_test.clj | 62 +++++++++++++++++ test/linear/handler/turso/protobuf_test.clj | 10 +++ test/linear/handler/turso/pull_test.clj | 1 - test/linear/usecase/database/pull_test.clj | 73 ++++++++++++++++++++ 16 files changed, 313 insertions(+), 9 deletions(-) create mode 100644 test/linear/handler/turso/long_poll_test.clj diff --git a/compose.yaml b/compose.yaml index d512cde..47e3fd7 100644 --- a/compose.yaml +++ b/compose.yaml @@ -27,6 +27,7 @@ services: JDBC_DATABASE_URL: jdbc:postgresql://postgres:5432/linear?user=postgres&password=postgres SLATEDB_OBJECT_STORE_URL: memory:/// REVISION_RETENTION_DAYS: + MAX_LONG_POLL_TIMEOUT_MS: LINEAR_DEPLOYMENT_TARGET: GCP_KMS_KEY_NAME: OIDC_ISSUER: diff --git a/docs/testing.md b/docs/testing.md index ff5b8c7..56b7df5 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -22,6 +22,7 @@ The [E2E scenario](../test/e2e/turso-sync.mjs) covers: - HTTP resource creation, empty bootstrap, rename, closure, and closed-source recovery; - incremental pull, local-write push, concurrent merge, and chunked pull; +- bounded long polling, repeated early responses, updates during waiting, and aborted-pull recovery; - lost-response retries, reconnect, SQL constraints, and DDL rollback; - pull-only bootstrap with push denied, and ungranted bootstrap denied. diff --git a/docs/turso-compatibility.md b/docs/turso-compatibility.md index 779cb15..a50160d 100644 --- a/docs/turso-compatibility.md +++ b/docs/turso-compatibility.md @@ -12,7 +12,8 @@ E2E scenarios, run by [CI](../.github/workflows/ci.yml) on Linux x64. | Retry after a lost push response | Supported; applied progress prevents duplicate SQL execution | | Atomic application data and sync progress | Supported; committed in the same revision | | SQL failure | Entire batch rolls back; COMMIT reports `TRANSACTION_ROLLED_BACK` | -| Query partial sync and long polling | Unsupported | +| Bounded long polling | Supported; official SDK early-response and reconnect scenarios are covered by E2E | +| Query partial sync | Unsupported | | MVCC and experimental remote writes | Outside the verified compatibility scope | ## Limitations @@ -25,6 +26,34 @@ E2E scenarios, run by [CI](../.github/workflows/ci.yml) on Linux x64. - Clients from earlier Linear versions need fresh local bootstrap because those versions discarded sync progress, which cannot be reconstructed. +## Long polling + +`long_poll_timeout_ms` applies when the client supplies `client_revision` and +omits `server_revision`. An available revision returns immediately. Otherwise, +Linear waits for at most the smaller of the requested duration and +`MAX_LONG_POLL_TIMEOUT_MS` (default **5000 ms**). Zero requests an immediate pull. +The limit bounds waiting, not the total time spent reading storage or sending +the response. A request above the limit receives a normal response, not an error. + +The five-second default follows the explicit value in the upstream v0.8.3 +[browser example](https://github.com/tursodatabase/turso/blob/v0.8.3/examples/javascript/sync-wasm-vite/index.html#L50-L55). +It is an operational limit, not a mobile-network connection guarantee. Configure +the deployment's HTTP timeouts with room for storage and response processing. + +The SlateDB adapter checks HEAD through fresh snapshots and sleeps for at most +100 ms between checks. Each check releases its snapshot and database lease before sleeping. +Waiting holds no PostgreSQL transaction and creates no subscriptions or detached +tasks. This uses the existing SlateDB binding without a notification registry +or application locks. Authorization and database state are checked again before +the final pull. + +The HTTP handler is synchronous. A disconnected request can keep waiting until +its bounded deadline; it does not cancel immediately on socket closure. +[Transport tests](../test/linear/handler/turso/long_poll_test.clj) repeatedly close +TCP connections during waits and verify that the waits finish. The official SDK +E2E checks capped empty responses, subsequent pulls, updates during waiting, and +recovery after an aborted request. + ## SDK 0.8.3 update The [upstream 0.8.2 to 0.8.3 comparison](https://github.com/tursodatabase/turso/compare/v0.8.2...v0.8.3) @@ -42,7 +71,6 @@ implementation or verification work, not passing claims for Linear: | [Python `test_partial_sync`](https://github.com/tursodatabase/turso/blob/v0.8.3/bindings/python/tests/test_database_sync.py) | Query bootstrap; page selection via `server_pages_selector` is implemented locally. | | [JavaScript partial-sync tests](https://github.com/tursodatabase/turso/blob/v0.8.3/bindings/javascript/sync/packages/native/promise.test.ts) | Partial-sync concurrency, segment size, and prefetch. | | [Python `test_pull_bytes_threshold`](https://github.com/tursodatabase/turso/blob/v0.8.3/bindings/python/tests/test_database_sync.py) | Upstream edge cases beyond the repeated selector-based pulls covered by E2E. | -| [`PullUpdatesReqProtoBody.long_poll_timeout_ms`](https://github.com/tursodatabase/turso/blob/v0.8.3/sync/engine/src/server_proto.rs) | Long-poll wake-up on a newly published revision. | Upstream includes a [CLI sync server](https://github.com/tursodatabase/turso/blob/v0.8.3/cli/sync_server.rs) with page-selector support. Its JavaScript partial-sync tests skip that server diff --git a/duct.edn b/duct.edn index 7aca0da..6f20c91 100644 --- a/duct.edn +++ b/duct.edn @@ -11,6 +11,9 @@ revision-retention-days {:env "REVISION_RETENTION_DAYS" :type :int :default 30} + max-long-poll-timeout-ms {:env "MAX_LONG_POLL_TIMEOUT_MS" + :type :int + :default 5000} sqlite-vfs-name {:env "SQLITE_VFS_NAME" :default "linear-sqlite"} deployment-target {:env "LINEAR_DEPLOYMENT_TARGET" @@ -56,7 +59,8 @@ :linear.usecase.core/database #ig/ref :linear.adapter.postgres/datasource :linear.usecase.core/revision-store #ig/ref :linear.adapter.slatedb.store/store :linear.usecase.core/evaluator #ig/ref :linear.adapter.sqlite.evaluator/evaluator - :linear.usecase.core/key-service #ig/ref :linear.adapter.crypto/key-service} + :linear.usecase.core/key-service #ig/ref :linear.adapter.crypto/key-service + :linear.usecase.core/max-long-poll-timeout-ms #ig/var max-long-poll-timeout-ms} :middleware-opts {:provider #ig/ref :duct.auth.jwks/provider :issuer #ig/var oidc-issuer diff --git a/src/linear/adapter/slatedb/store.clj b/src/linear/adapter/slatedb/store.clj index 9da731f..8a1d8d0 100644 --- a/src/linear/adapter/slatedb/store.clj +++ b/src/linear/adapter/slatedb/store.clj @@ -125,7 +125,21 @@ revisions/RevisionWritable (-publish-next! [store revision database] - (publish! store revision database false))) + (publish! store revision database false)) + + revisions/RevisionAwaitable + (-await-revision [store {:keys [database revision-id timeout-ms]}] + (let [deadline (+ (System/nanoTime) (* timeout-ms 1000000))] + (loop [] + (let [current-revision-id (with-consistent-view store database + #(revisions/revision-id (revisions/head %))) + remaining (- deadline (System/nanoTime))] + (cond + (not= current-revision-id revision-id) true + (not (pos? remaining)) false + :else (do + (Thread/sleep (long (min 100 (max 1 (quot remaining 1000000))))) + (recur)))))))) (defmethod integrant/init-key :linear.adapter.slatedb.store/store [_ options] diff --git a/src/linear/handler/turso/protobuf.clj b/src/linear/handler/turso/protobuf.clj index 1972e15..4ff9e49 100644 --- a/src/linear/handler/turso/protobuf.clj +++ b/src/linear/handler/turso/protobuf.clj @@ -84,6 +84,8 @@ (malformed "Pull client revision has an invalid wire type")) 4 (if (= 0 wire-type) (let [[value end] (read-varint data next-offset)] + (when-not (<= 0 value 0xffffffff) + (malformed "Pull long poll timeout must be a uint32")) (recur (long end) (assoc pull :long-poll-timeout-ms value))) (malformed "Pull long poll timeout has an invalid wire type")) 5 (if (= 2 wire-type) diff --git a/src/linear/handler/turso/pull.clj b/src/linear/handler/turso/pull.clj index 93838cc..171f706 100644 --- a/src/linear/handler/turso/pull.clj +++ b/src/linear/handler/turso/pull.clj @@ -31,7 +31,6 @@ (defn- validate-options [pull] (when (or (not (zero? (get pull :encoding 0))) (not (zero? (get pull :stream-kind 0))) - (not (zero? (get pull :long-poll-timeout-ms 0))) (seq (:server-query-selector pull)) (seq (:client-pages pull))) (throw (ex-info "Pull option is not supported" @@ -45,6 +44,7 @@ _ (validate-options pull) result (database/pull context {:server-revision (not-empty (:server-revision pull)) :client-revision (not-empty (:client-revision pull)) + :long-poll-timeout-ms (get pull :long-poll-timeout-ms 0) :page-ids (protobuf/decode-page-selector (:server-pages-selector pull)) :actor (:sub identity) diff --git a/src/linear/usecase/core.clj b/src/linear/usecase/core.clj index 82fc53e..6dd010f 100644 --- a/src/linear/usecase/core.clj +++ b/src/linear/usecase/core.clj @@ -28,5 +28,15 @@ [{::keys [revision-store]}] revision-store) +(defn revision-awaitable + [{::keys [revision-store]}] + revision-store) + +(defn max-long-poll-timeout-ms + {:malli/schema [:-> :map pos-int?]} + [{::keys [max-long-poll-timeout-ms] + :or {max-long-poll-timeout-ms 5000}}] + max-long-poll-timeout-ms) + (defn checkables [context] [(transactable context) (evaluator context)]) diff --git a/src/linear/usecase/database.clj b/src/linear/usecase/database.clj index e56a5bf..560c6f6 100644 --- a/src/linear/usecase/database.clj +++ b/src/linear/usecase/database.clj @@ -191,13 +191,30 @@ error)) (throw error))))) -(defn pull +(defn- pull-once [context {:keys [actor database-id] :as pull-options}] (with-database [database context {:database-id database-id :actor actor :permission :pull :read-only true}] - (model/pull database pull-options))) + [database (model/pull database pull-options)])) + +(defn pull + [context {:keys [client-revision server-revision long-poll-timeout-ms] + :or {long-poll-timeout-ms 0} + :as pull-options}] + (let [[database result] (pull-once context pull-options)] + (if (and client-revision (nil? server-revision) + (pos? long-poll-timeout-ms) + (= client-revision (:server-revision result))) + (do + (revisions/await-revision (core/revision-awaitable context) + {:database (select-keys database [:id :vault]) + :revision-id client-revision + :timeout-ms (min long-poll-timeout-ms + (core/max-long-poll-timeout-ms context))}) + (second (pull-once context pull-options))) + result))) (defn sync-progress [context {:keys [actor database-id] :as arg-map}] diff --git a/src/linear/usecase/database/revisions.clj b/src/linear/usecase/database/revisions.clj index e1930a9..7e48bc9 100644 --- a/src/linear/usecase/database/revisions.clj +++ b/src/linear/usecase/database/revisions.clj @@ -91,6 +91,19 @@ (defprotocol RevisionWritable (-publish-next! [writer revision database])) +(defprotocol RevisionAwaitable + (-await-revision [reader arg-map])) + +(defn await-revision + {:malli/schema [:-> [:fn #(satisfies? RevisionAwaitable %)] + [:map + [:database [:map [:id :string] [:vault [:map [:keychain :any]]]]] + [:revision-id ::revision-id] + [:timeout-ms pos-int?]] + :boolean]} + [reader arg-map] + (-await-revision reader arg-map)) + (defprotocol Initializable (-initialize! [writer arg-map])) diff --git a/test/e2e/turso-sync.mjs b/test/e2e/turso-sync.mjs index 766364f..d8e35f2 100644 --- a/test/e2e/turso-sync.mjs +++ b/test/e2e/turso-sync.mjs @@ -109,6 +109,75 @@ try { await readerClient.exec("UPDATE t SET value = 'reader-local' WHERE id = 1"); await assert.rejects(() => readerClient.push(), /403/); + const shortPollClient = await connectClient("long-poll-short", { longPollTimeoutMs: 100 }); + const shortPollStart = performance.now(); + assert.equal(await shortPollClient.pull(), false); + const shortPollElapsed = performance.now() - shortPollStart; + assert(shortPollElapsed >= 90 && shortPollElapsed < 4000, + `requested short wait was not respected: ${shortPollElapsed} ms`); + + // The official SDK must accept a normal empty pull before its requested + // timeout, then keep pulling from the same revision. + const cappedClient = await connectClient("long-poll-capped", { + longPollTimeoutMs: 0xffffffff + }); + for (let attempt = 0; attempt < 2; attempt += 1) { + const start = performance.now(); + assert.equal(await cappedClient.pull(), false); + const elapsed = performance.now() - start; + assert(elapsed >= 4900, `server cap returned too early: ${elapsed} ms`); + assert(elapsed < 9000, `server cap did not bound waiting: ${elapsed} ms`); + } + + let observePull; + let abortPull; + const longPollClient = await connectClient("long-poll", { + longPollTimeoutMs: 5000, + fetch: async (input, init) => { + if (String(input).endsWith("/pull-updates") && observePull) { + const observed = observePull; + observePull = undefined; + observed(); + if (abortPull) { + return timedFetch(input, { ...init, signal: abortPull.signal }); + } + } + return timedFetch(input, init); + } + }); + const waiting = new Promise(resolve => { observePull = resolve; }); + let completed = false; + const pulling = longPollClient.pull().then(changed => { + completed = true; + return changed; + }); + await waiting; + await new Promise(resolve => setTimeout(resolve, 100)); + assert.equal(completed, false); + const wakeStart = performance.now(); + await clientA.exec("UPDATE t SET value = 'long-poll-update' WHERE id = 1"); + await clientA.push(); + assert.equal(await pulling, true); + assert(performance.now() - wakeStart < 4000, "published update did not wake the pull before its deadline"); + assert.deepEqual(await rows(longPollClient, "SELECT value FROM t WHERE id = 1"), [ + { value: "long-poll-update" } + ]); + + const disconnecting = new Promise(resolve => { observePull = resolve; }); + abortPull = new AbortController(); + const interrupted = assert.rejects(() => longPollClient.pull()); + await disconnecting; + abortPull.abort(); + await interrupted; + abortPull = undefined; + await clientA.exec("UPDATE t SET value = 'from-js' WHERE id = 1"); + await clientA.push(); + assert.equal(await longPollClient.pull(), true); + assert.deepEqual(await rows(longPollClient, "SELECT value FROM t WHERE id = 1"), [ + { value: "from-js" } + ]); + await clientB.pull(); + await assert.rejects( () => connectClient("ungranted", { authToken: ungrantedToken }), /403/ diff --git a/test/linear/e2e.clj b/test/linear/e2e.clj index 868d3d5..c2d526d 100644 --- a/test/linear/e2e.clj +++ b/test/linear/e2e.clj @@ -85,7 +85,8 @@ system (test/run {:config (config) :keys #{:duct.server.http/jetty} :profiles [:test :main] - :vars (assoc (jwt/oidc-vars fixture) 'port 0)})] + :vars (assoc (jwt/oidc-vars fixture) + 'port 0 'max-long-poll-timeout-ms 5000)})] (try (run-client! (server-url system) {"LINEAR_E2E_MANAGER_TOKEN" (jwt/access-token fixture {:subject manager}) diff --git a/test/linear/handler/turso/long_poll_test.clj b/test/linear/handler/turso/long_poll_test.clj new file mode 100644 index 0000000..e7c44a3 --- /dev/null +++ b/test/linear/handler/turso/long_poll_test.clj @@ -0,0 +1,62 @@ +(ns linear.handler.turso.long-poll-test + (:require + [clojure.test :refer [deftest is]] + [duct.test :refer [with-system]] + [linear.adapter.postgres] + [linear.adapter.slatedb.store] + [linear.handler.turso.pull :as pull] + [linear.test :refer [run]] + [linear.test-data.postgres :as postgres-data] + [linear.test-data.slatedb :as slatedb-data] + [linear.usecase.core :as core] + [linear.usecase.database.revisions :as revisions] + [ring.adapter.jetty :as jetty]) + (:import + (java.net Socket) + (java.nio.charset StandardCharsets) + (java.util.concurrent LinkedBlockingQueue TimeUnit))) + +(deftest ^:integration disconnected-http-pulls-release-their-revision-waits + (with-system [system (run {:keys [:duct.database/sql :duct.migrator/ragtime + :linear.adapter.slatedb.store/store]})] + (let [datasource (:duct.database.sql/hikaricp system) + store (:linear.adapter.slatedb.store/store system) + {:keys [database-id key-protection keychain]} + (postgres-data/create-database! {:datasource datasource}) + events (LinkedBlockingQueue.) + reader (reify + revisions/ConsistentReadable + (-read-consistently [_ f database] + (revisions/-read-consistently store f database)) + revisions/RevisionAwaitable + (-await-revision [_ args] + (.offer events :entered) + (try + (revisions/await-revision store args) + (finally (.offer events :released))))) + handler (pull/handler {::core/database datasource ::core/key-service key-protection + ::core/revision-store reader ::core/max-long-poll-timeout-ms 100}) + server (jetty/run-jetty #(handler (assoc % :identity {:sub "actor-1"} + :path-params {:id database-id})) + {:port 0 :join? false}) + port (.getLocalPort (first (.getConnectors server))) + body (byte-array (map unchecked-byte [0x1a 6 0x72 0x2d 0x72 0x6f 0x6f 0x74 0x20 0x88 0x27]))] + (try + (slatedb-data/store-root! {:store store :database-id database-id :keychain keychain :pages {}}) + (dotimes [_ 10] + (with-open [socket (Socket. "127.0.0.1" port)] + (let [output (.getOutputStream socket)] + (.write output (.getBytes (str "POST /pull-updates HTTP/1.1\r\nHost: localhost\r\n" + "Connection: close\r\nContent-Length: " (alength body) "\r\n\r\n") + StandardCharsets/UTF_8)) + (.write output body) + (.flush output) + (is (= :entered (.poll events 2 TimeUnit/SECONDS))) + (.setSoLinger socket true 0))) + ;; This handler does not observe disconnects during a pull. The + ;; bounded revision wait must finish without a response write. + (is (= :released (.poll events 2 TimeUnit/SECONDS)))) + (is (nil? (.poll events))) + (finally + (.stop server) + (.join server)))))) diff --git a/test/linear/handler/turso/protobuf_test.clj b/test/linear/handler/turso/protobuf_test.clj index 00fb3dc..a85a2c2 100644 --- a/test/linear/handler/turso/protobuf_test.clj +++ b/test/linear/handler/turso/protobuf_test.clj @@ -81,6 +81,16 @@ 0x12 0x02 0x01 0x02)) (seq (.toByteArray output)))))) +(deftest long-poll-timeout-is-an-unsigned-32-bit-value + (is (= 0xffffffff + (:long-poll-timeout-ms + (protobuf/decode-pull (byte-array (map unchecked-byte [0x20 0xff 0xff 0xff 0xff 0x0f])))))) + (doseq [value [[0x20 0x80 0x80 0x80 0x80 0x10] + [0x20 0xff 0xff 0xff 0xff 0xff 0xff 0xff 0xff 0xff 0x01]]] + (is (= ::protobuf/malformed-protobuf + (:reason (test/catch-ex-data + #(protobuf/decode-pull (byte-array (map unchecked-byte value))))))))) + (deftest protobuf-parser-rejects-truncated-length-delimited-fields (testing "syntax errors are codec errors, not semantic validation" (is (thrown? Exception diff --git a/test/linear/handler/turso/pull_test.clj b/test/linear/handler/turso/pull_test.clj index 9a91633..7b107af 100644 --- a/test/linear/handler/turso/pull_test.clj +++ b/test/linear/handler/turso/pull_test.clj @@ -21,7 +21,6 @@ (deftest unsupported-pull-options-are-rejected-before-database-resolution (doseq [[option bytes] [[:encoding [0x08 0x01]] [:stream-kind [0x40 0x01]] - [:long-poll-timeout-ms [0x20 0x01]] [:server-query-selector [0x3a 0x01 0x71]] [:client-pages [0x32 0x01 0x01]]]] (is (= 400 (:status ((pull/handler {}) {:body (byte-array bytes)}))) diff --git a/test/linear/usecase/database/pull_test.clj b/test/linear/usecase/database/pull_test.clj index a9b8dab..ea43b9c 100644 --- a/test/linear/usecase/database/pull_test.clj +++ b/test/linear/usecase/database/pull_test.clj @@ -99,3 +99,76 @@ (is (thrown? clojure.lang.ExceptionInfo (database/pull ctx pull-args))) (restore-fn ctx (assoc args :token token)) (is (= (seq page) (seq (get-in (database/pull ctx pull-args) [:pages 1]))))))) + +(deftest ^:integration long-poll-caps-waiting-and-closes-read-scopes-before-awaiting + (with-system [system (run {:keys [:duct.database/sql :duct.migrator/ragtime + :linear.adapter.slatedb.store/store]})] + (let [datasource (:duct.database.sql/hikaricp system) + store (:linear.adapter.slatedb.store/store system) + {:keys [database-id key-protection keychain]} + (postgres-data/create-database! {:datasource datasource}) + connection (atom nil) + viewing? (atom false) + waits (atom []) + transactable (reify transaction/Transactable + (-transact [_ f options] + (transaction/-transact datasource + (fn [tx] (reset! connection tx) (f tx)) options))) + reader (reify + revisions/ConsistentReadable + (-read-consistently [_ f database] + (revisions/-read-consistently store + (fn [view] + (reset! viewing? true) + (try (f view) (finally (reset! viewing? false)))) + database)) + revisions/RevisionAwaitable + (-await-revision [_ args] + (is (false? @viewing?)) + (is (.isClosed ^java.sql.Connection @connection)) + (swap! waits conj (select-keys args [:revision-id :timeout-ms])) + false)) + context {::core/database transactable ::core/key-service key-protection + ::core/revision-store reader ::core/max-long-poll-timeout-ms 40} + args {:actor "actor-1" :database-id database-id}] + (slatedb-data/store-root! {:store store :database-id database-id :keychain keychain :pages {}}) + (doseq [options [{:long-poll-timeout-ms 5000} + {:client-revision "r-root" :long-poll-timeout-ms 0} + {:client-revision "r-root" :server-revision "r-root" :long-poll-timeout-ms 5000}]] + (is (= "r-root" (:server-revision (database/pull context (merge args options)))))) + (is (empty? @waits)) + (doseq [timeout [10 5000 0xffffffff]] + (is (= {} (:pages (database/pull context (assoc args :client-revision "r-root" + :long-poll-timeout-ms timeout)))))) + (is (= [{:revision-id "r-root" :timeout-ms 10} + {:revision-id "r-root" :timeout-ms 40} + {:revision-id "r-root" :timeout-ms 40}] @waits)) + (revisions/publish-next! store + {:database {:id database-id :vault {:keychain keychain}} + :revision {:revision-id "r-next" :parent "r-root" :database-page-count 1 :pages {1 (byte-array [7])}}}) + (let [result (database/pull context (assoc args :client-revision "r-root" :long-poll-timeout-ms 5000))] + (is (= "r-next" (:server-revision result))) + (is (= {1 '(7)} (update-vals (:pages result) seq))) + (is (= 3 (count @waits))))))) + +(deftest ^:integration long-poll-rechecks-database-access-after-waiting + (with-system [system (run {:keys [:duct.database/sql :duct.migrator/ragtime + :linear.adapter.slatedb.store/store]})] + (let [datasource (:duct.database.sql/hikaricp system) + store (:linear.adapter.slatedb.store/store system) + {:keys [database-id key-protection keychain]} + (postgres-data/create-database! {:datasource datasource}) + context {::core/database datasource ::core/key-service key-protection ::core/revision-store store} + args {:actor "actor-1" :database-id database-id} + reader (reify + revisions/ConsistentReadable + (-read-consistently [_ f database] (revisions/-read-consistently store f database)) + revisions/RevisionAwaitable + (-await-revision [_ _] + (database/close! context args) + false))] + (slatedb-data/store-root! {:store store :database-id database-id :keychain keychain :pages {}}) + (is (= ::database/database-not-found + (:reason (linear.test/catch-ex-data + #(database/pull (assoc context ::core/revision-store reader) + (assoc args :client-revision "r-root" :long-poll-timeout-ms 100)))))))))