yujiyokoo · Nov 13, 2026 · Nov 3, 2026 · Nov 3, 2026 · Nov 3, 2026
@@ -150,6 +150,20 @@ With this form, a new `Book` has `author` set to `None` until you set it. A bare
 
 `Author` must be defined before `Book`, because the decorator evaluates `"Author"` in the hint when `Book`'s class statement runs.
 
+## Use it from several threads
+
+`encrypt`, `decrypt`, and `query` are synchronous. There is no async form yet.
+
+Each call releases the GIL while the encryption runs. The GIL is CPython's global interpreter lock. So other Python threads keep running during a call, and calls from several threads run at the same time. Within one call, the requests to CipherStash go out one at a time.
+
+A process has one connection to CipherStash. The first call of any `Cipher` makes it, reads the credentials, and loads the default keyset. Every `Cipher` in the process then uses that connection. If you change the `CS_*` variables or log in again, restart the process.
+
+Make one `Cipher` for each model, and share it between threads with no lock. Do not make a new `Cipher` for each web request. Making one checks the model and the credentials. When no access key is set in the environment, it also reads your CipherStash profile from disk.
+
+If you press Ctrl-C during a call, the call finishes first. Python then raises `KeyboardInterrupt`, and the `Cipher` still works. A call cannot be cancelled.
+
+Each request to ZeroKMS, CipherStash's key service, times out after 10 seconds. A request for an access token times out after 30 seconds. The SDK makes one on the first call when no fresh token is stored, and again shortly before the token expires. A call that makes several requests can take longer in total. You cannot change these timeouts yet.
+
 ## Run the tests
 
 From `languages/python`:

@@ -9,13 +9,14 @@
 //! Python threads run inside the engine at the same time.
 //!
 //! The client is built over a ZeroKMS connection that counts every request
-//! it sends. The count is for the tests: the spec's request rules (one key
-//! request per 500 values, one keyset load per process) are observed
-//! through it, not assumed.
+//! it sends, by endpoint. The counts are for the tests: the spec's request
+//! rules (one key request per 500 values, one keyset load per process) are
+//! observed through them, not assumed.
 
+use std::collections::BTreeMap;
 use std::future::Future;
 use std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::{Mutex, OnceLock};
+use std::sync::{Mutex, MutexGuard, OnceLock};
 
 use stack_auth::AutoStrategy;
 use stack_encrypt::kms::{
@@ -45,7 +46,17 @@ static CLIENT: OnceLock<Client> = OnceLock::new();
 // fixes the credentials and calls again.
 static BUILDING: Mutex<()> = Mutex::new(());
 static BUILDS: AtomicU64 = AtomicU64::new(0);
-static REQUESTS: AtomicU64 = AtomicU64::new(0);
+// Every request sent, by its ZeroKMS endpoint, such as `load-keyset`.
+static REQUESTS: Mutex<BTreeMap<&'static str, u64>> = Mutex::new(BTreeMap::new());
+
+/// `mutex`, locked. A thread that panicked while it held a lock here left
+/// nothing half-written: a build is either cached or not, and a count is
+/// one addition. So a poisoned lock is used as it is.
+fn locked<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
+    mutex
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+}
 
 /// The shared client, built on the first call.
 ///
@@ -60,9 +71,7 @@ pub fn shared() -> Result<&'static Client, Refusal> {
     if let Some(client) = CLIENT.get() {
         return Ok(client);
     }
-    let _building = BUILDING
-        .lock()
-        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    let _building = locked(&BUILDING);
     // Again under the lock: a switch can finish while this thread waits for
     // it, and the live client must then not be built.
     #[cfg(feature = "deterministic-kms")]
@@ -145,9 +154,7 @@ pub fn use_deterministic(seed: [u8; 32]) -> Result<(), Refusal> {
     };
     // The lock `shared` builds under: a live build in flight on another
     // thread finishes before this checks, so the two never both exist.
-    let _building = BUILDING
-        .lock()
-        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    let _building = locked(&BUILDING);
     if CLIENT.get().is_some() || DETERMINISTIC.get().is_some() {
         return Err(refused(
             "the process already has a client; call this before any operation".to_owned(),
@@ -184,7 +191,15 @@ pub fn build_count() -> u64 {
 /// How many requests the shared client has sent to ZeroKMS in this process,
 /// the keyset load included. For the tests.
 pub fn request_count() -> u64 {
-    REQUESTS.load(Ordering::Relaxed)
+    requests().values().sum()
+}
+
+/// The requests the shared client has sent to ZeroKMS in this process, by
+/// endpoint. A request counts when it is sent, whether or not it succeeds.
+/// For the tests: `load-keyset` is 1 after the first operation, however
+/// many threads made it at once.
+pub fn requests() -> BTreeMap<&'static str, u64> {
+    locked(&REQUESTS).clone()
 }
 
 impl Client {
@@ -295,7 +310,9 @@ impl ZeroKMSConnection for Counting {
         request: Request,
         access_token: &str,
     ) -> Result<Request::Response, ViturRequestError> {
-        REQUESTS.fetch_add(1, Ordering::Relaxed);
+        // Before the send: a request that fails is still a request, which
+        // the tests of a refused call rely on.
+        *locked(&REQUESTS).entry(Request::ENDPOINT).or_default() += 1;
         self.0.send(request, access_token).await
     }
 

@@ -173,6 +173,14 @@ pub fn request_count() -> u64 {
     stackencrypt_core::client::request_count()
 }
 
+/// The requests the shared client has sent to ZeroKMS, by endpoint, such
+/// as `load-keyset`. For the tests: threads that make their first call at
+/// once load the keyset once.
+#[pyfunction]
+pub fn requests() -> std::collections::BTreeMap<&'static str, u64> {
+    stackencrypt_core::client::requests()
+}
+
 /// `value` read as `field`'s Python type and checked against the range of
 /// its kind, as `encrypt` and `query` both take it. A refusal names the
 /// field, with the kind as its reason.

@@ -114,6 +114,7 @@ fn _engine(module: &Bound<'_, PyModule>) -> PyResult<()> {
     module.add_function(wrap_pyfunction!(query::query, module)?)?;
     module.add_function(wrap_pyfunction!(encrypt::client_builds, module)?)?;
     module.add_function(wrap_pyfunction!(encrypt::request_count, module)?)?;
+    module.add_function(wrap_pyfunction!(encrypt::requests, module)?)?;
     #[cfg(feature = "gate-probe")]
     module.add_function(wrap_pyfunction!(_probe_detached_sleep, module)?)?;
     #[cfg(feature = "deterministic-kms")]

@@ -35,6 +35,8 @@ features = ["pyo3/extension-module"]
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
+# tests/_sync_contract.py is shared with tests_deterministic/.
+pythonpath = ["tests"]
 # `live` marks the tests that need credentials and a database. They skip
 # without them; `mise run test:live` runs only them and fails on a skip.
 markers = ["live: needs CipherStash credentials, and some a PostgreSQL with EQL v4"]

@@ -45,3 +45,7 @@ def client_builds() -> int:
 
 def request_count() -> int:
     """How many requests the shared client has sent to ZeroKMS. For tests."""
+
+def requests() -> Dict[str, int]:
+    """The requests the shared client has sent to ZeroKMS, by endpoint,
+    such as ``load-keyset``. For tests."""
@@ -0,0 +1,19 @@
+"""Child process of `test_concurrency_live.py`: the threads make the first
+call of this process together. Prints the counts as JSON."""
+
+import json
+
+from _sync_contract import User, encrypt_from_threads
+
+from stackencrypt import Cipher, _engine
+
+assert _engine.client_builds() == 0
+encrypt_from_threads(Cipher(User), per_thread=1)
+print(
+    json.dumps(
+        {
+            "builds": _engine.client_builds(),
+            "keyset_loads": _engine.requests().get("load-keyset", 0),
+        }
+    )
+)
@@ -0,0 +1,68 @@
+"""Child process of `test_concurrency_live.py`: SIGINT during a long
+`encrypt`. Prints what happened as JSON.
+
+A reference time is measured first, as the gate test knows its call takes
+one second. It is the fastest of three calls, so a slow one cannot raise
+it. The call starts at 5000 values and doubles until the reference is long
+enough to send a signal into. The signal is sent a quarter of the way into
+one more call. A call cut short would raise at that quarter. A call that
+waits raises after it returns, near the reference time.
+"""
+
+import json
+import time
+
+from _sync_contract import User, send_sigint_later, user, wait_for_the_interrupt
+
+from stackencrypt import Cipher
+
+LONG_ENOUGH = 0.4
+MOST_VALUES = 80_000
+
+
+def timed(call):
+    started = time.monotonic()
+    call()
+    return time.monotonic() - started
+
+
+cipher = Cipher(User)
+# Load the keyset first, so the timed calls are the values alone.
+cipher.encrypt([user(0)])
+
+values = 5000
+while True:
+    models = [user(n) for n in range(values)]
+    reference = min(timed(lambda: cipher.encrypt(models)) for _ in range(3))
+    if reference >= LONG_ENOUGH or values >= MOST_VALUES:
+        break
+    values *= 2
+
+interrupts = 0
+arrived = None
+rows = None
+timer = send_sigint_later(reference / 4)
+started = time.monotonic()
+try:
+    rows = cipher.encrypt(models)
+    wait_for_the_interrupt()
+except KeyboardInterrupt:
+    interrupts += 1
+    arrived = time.monotonic() - started
+timer.join()
+
+# Python checks for signals many times in the code below. A second
+# interrupt would raise in it, and the process would exit with an error.
+after = cipher.encrypt([user(1)])
+print(
+    json.dumps(
+        {
+            "values": values,
+            "reference": reference,
+            "arrived": arrived,
+            "interrupts": interrupts,
+            "rows": None if rows is None else len(rows),
+            "after": len(after),
+        }
+    )
+)
@@ -0,0 +1,73 @@
+"""The parts of the sync contract tests that more than one file uses: the
+model, the threads, and the interrupt.
+
+The users are `tests/test_concurrency_live.py`, the deterministic copy in
+`tests_deterministic/test_concurrency.py`, the two child scripts
+`_first_call.py` and `_interrupt.py`, and the gate test
+`test_gate_keyboard_interrupt.py`. Not a test file: pytest collects
+nothing here.
+"""
+
+import dataclasses
+import os
+import signal
+import threading
+import time
+from typing import Annotated, Dict, List
+
+import stackencrypt
+from stackencrypt import Cipher, Passthrough, TextEq
+
+THREADS = 8
+PER_THREAD = 100
+
+
+@stackencrypt.model("users")
+@dataclasses.dataclass
+class User:
+    id: Annotated[int, Passthrough]
+    email: Annotated[str, TextEq]
+
+
+def user(n: int) -> User:
+    return User(id=n, email=f"user{n}@example.com")
+
+
+def encrypt_from_threads(cipher: Cipher, per_thread: int) -> List[Dict[str, object]]:
+    """Release ``THREADS`` threads at once, each to encrypt ``per_thread``
+    users one call at a time. Returns every row, in user order. Raises if
+    any call failed."""
+    sealed: List[List[Dict[str, object]]] = [[] for _ in range(THREADS)]
+    failures: List[str] = []
+    start = threading.Barrier(THREADS)
+
+    def encrypt_many(t: int) -> None:
+        start.wait()
+        try:
+            for n in range(per_thread):
+                sealed[t].extend(cipher.encrypt([user(t * per_thread + n)]))
+        except Exception as error:
+            failures.append(repr(error))
+
+    threads = [threading.Thread(target=encrypt_many, args=(t,)) for t in range(THREADS)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+    assert failures == [], failures
+    return [row for rows in sealed for row in rows]
+
+
+def send_sigint_later(seconds: float) -> threading.Timer:
+    """Send SIGINT to this process after ``seconds``."""
+    timer = threading.Timer(seconds, os.kill, (os.getpid(), signal.SIGINT))
+    timer.start()
+    return timer
+
+
+def wait_for_the_interrupt() -> None:
+    """Python raises a pending interrupt at its next check after a call. The
+    loop gives it that check, and ends if it never comes."""
+    deadline = time.monotonic() + 5
+    while time.monotonic() < deadline:
+        pass
@@ -0,0 +1,72 @@
+"""The sync contract from the spec in #1341, held against ZeroKMS: one engine
+client per process, shared by every thread, and a ``KeyboardInterrupt``
+that waits for the call.
+
+These tests need the four CipherStash credentials. Without them every test
+here skips. The first-call test and the interrupt test each run a script in
+a new Python process: the client of this one is already built, and the
+interrupt is for the whole process.
+"""
+
+import json
+import os
+import pathlib
+import subprocess
+import sys
+
+import pytest
+from _sync_contract import PER_THREAD, THREADS, User, encrypt_from_threads, user
+
+from stackencrypt import Cipher
+
+CREDENTIALS = ("CS_CLIENT_ACCESS_KEY", "CS_WORKSPACE_CRN", "CS_CLIENT_ID", "CS_CLIENT_KEY")
+
+pytestmark = [
+    pytest.mark.live,
+    pytest.mark.skipif(
+        not all(os.environ.get(name) for name in CREDENTIALS),
+        reason="needs CS_CLIENT_ACCESS_KEY, CS_WORKSPACE_CRN, CS_CLIENT_ID and CS_CLIENT_KEY",
+    ),
+]
+
+
+def run_child(script):
+    """Run ``script``, beside this file, in a new Python process with this
+    one's environment. Returns the JSON it printed last."""
+    done = subprocess.run(
+        [sys.executable, str(pathlib.Path(__file__).with_name(script))],
+        capture_output=True,
+        text=True,
+        timeout=120,
+    )
+    assert done.returncode == 0, done.stderr
+    return json.loads(done.stdout.strip().splitlines()[-1])
+
+
+def test_threads_making_their_first_call_at_once_build_one_client():
+    assert run_child("_first_call.py") == {"builds": 1, "keyset_loads": 1}
+
+
+def test_threads_encrypt_at_once_and_every_row_opens():
+    cipher = Cipher(User)
+
+    rows = encrypt_from_threads(cipher, PER_THREAD)
+
+    assert cipher.decrypt(rows) == [user(n) for n in range(THREADS * PER_THREAD)]
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="SIGINT is sent with os.kill")
+def test_an_interrupt_during_encrypt_waits_for_it_and_the_cipher_still_works():
+    result = run_child("_interrupt.py")
+
+    # The script grows the call until it takes 0.4 seconds, up to 80,000
+    # values. Below 0.2, there is too little time to send a signal into it.
+    assert result["reference"] >= 0.2, result
+    assert result["interrupts"] == 1, result
+    # A call cut short raises at a quarter of the reference. A call that
+    # waits raises after it returns, near the reference time.
+    assert result["arrived"] >= result["reference"] / 2, result
+    # All the rows, or the interrupt came before the assignment and left
+    # nothing: never part of a result.
+    assert result["rows"] in (None, result["values"]), result
+    assert result["after"] == 1, result
@@ -96,8 +96,14 @@ def test_1001_one_field_models_make_three_key_requests():
 
 def test_two_ciphers_share_one_engine_client():
     Cipher(User).encrypt([User(email="a@example.com", age=1)])
+    # Read after the first call: an earlier test may have loaded the keyset
+    # more than once, if a load failed and the next call loaded it again.
+    loads = _engine.requests().get("load-keyset")
+
     Cipher(Contact).encrypt([Contact(email="b@example.com")])
+
     assert _engine.client_builds() == 1
+    assert _engine.requests().get("load-keyset") == loads
 
 
 @stackencrypt.model("prices")