Skip to content

Storage API Reference

Pluggable storage backends for persistent and in-memory state management.

Durability ledger: the storage protocol every backend implements.

Claim dataclass

A claim attempt's outcome. result is meaningful only when DONE.

Source code in src/istos/consistency/storage.py 
27
28
29
30
31
@dataclass(frozen=True)
class Claim:
    """A claim attempt's outcome. ``result`` is meaningful only when DONE."""
    state: ClaimState
    result: Any = None

ClaimState

Bases: str, Enum

Outcome of :meth:StoragePlugin.claim_processed.

  • CLAIMED: this caller owns the key and must execute the work.
  • IN_FLIGHT: another caller owns it and has not finished — do not execute.
  • DONE: the work already ran; :attr:Claim.result is what it returned.
Source code in src/istos/consistency/storage.py 
15
16
17
18
19
20
21
22
23
24
class ClaimState(str, Enum):
    """Outcome of :meth:`StoragePlugin.claim_processed`.

    - CLAIMED:   this caller owns the key and must execute the work.
    - IN_FLIGHT: another caller owns it and has not finished — do **not** execute.
    - DONE:      the work already ran; :attr:`Claim.result` is what it returned.
    """
    CLAIMED = "claimed"
    IN_FLIGHT = "in_flight"
    DONE = "done"

Durability

Bases: str, Enum

Delivery semantics for handler execution.

  • AT_MOST_ONCE: Fire-and-forget. No logging, no dedup. Fastest.
  • AT_LEAST_ONCE: Logs every call to event_log. Retries may cause duplicates.
  • EXACTLY_ONCE: Logs + idempotency. The key is claimed before the handler runs, so a concurrent duplicate cannot execute alongside it; once the work finishes, later duplicates return the cached result.
Source code in src/istos/consistency/storage.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
class Durability(str, Enum):
    """
    Delivery semantics for handler execution.

    - AT_MOST_ONCE:  Fire-and-forget. No logging, no dedup. Fastest.
    - AT_LEAST_ONCE: Logs every call to event_log. Retries may cause duplicates.
    - EXACTLY_ONCE:  Logs + idempotency. The key is claimed *before* the handler
      runs, so a concurrent duplicate cannot execute alongside it; once the work
      finishes, later duplicates return the cached result.
    """
    AT_MOST_ONCE = "at_most_once"
    AT_LEAST_ONCE = "at_least_once"
    EXACTLY_ONCE = "exactly_once"

InMemoryStoragePlugin

Thread-safe in-memory storage with full durability support. Data is lost on restart — use for testing and development.

Source code in src/istos/consistency/storage.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
class InMemoryStoragePlugin:
    """
    Thread-safe in-memory storage with full durability support.
    Data is lost on restart — use for testing and development.
    """
    def __init__(self):
        self._store: dict[str, Any] = {}
        self._event_log: dict[str, List[dict]] = {}
        self._processed: dict[str, _Record] = {}
        self._lock = asyncio.Lock()

    # ---- Core key-value ----

    async def put(self, key: str, value: Any) -> None:
        async with self._lock:
            self._store[key] = value

    async def get(self, key: str) -> Optional[Any]:
        async with self._lock:
            return self._store.get(key)

    async def delete(self, key: str) -> None:
        async with self._lock:
            if key in self._store:
                del self._store[key]

    # ---- Event log ----

    async def log(self, key: str, value: Any, idempotency_key: Optional[str] = None) -> None:
        async with self._lock:
            # Skip only a finished key — a claim is still in flight, and its
            # event belongs in the log.
            rec = self._processed.get(idempotency_key) if idempotency_key else None
            if rec is not None and rec.done:
                return
            if key not in self._event_log:
                self._event_log[key] = []
            self._event_log[key].append({
                "value": value,
                "timestamp": time.time(),
                "idempotency_key": idempotency_key,
            })

    async def get_log(self, key: str, limit: int = 100) -> List[Any]:
        async with self._lock:
            entries = self._event_log.get(key, [])
            return list(reversed(entries[-limit:]))

    # ---- Idempotency ----

    async def claim_processed(
        self, idempotency_key: str, *, lease_s: float = DEFAULT_CLAIM_LEASE_S
    ) -> Claim:
        async with self._lock:
            rec = self._processed.get(idempotency_key)
            if rec is not None:
                if rec.done:
                    return Claim(ClaimState.DONE, rec.result)
                if rec.lease_until > time.monotonic():
                    return Claim(ClaimState.IN_FLIGHT)
                # Lease lapsed — whoever held it died mid-handler. Take over.
            self._processed[idempotency_key] = _Record(
                done=False, lease_until=time.monotonic() + lease_s
            )
            return Claim(ClaimState.CLAIMED)

    async def release_claim(self, idempotency_key: str) -> None:
        async with self._lock:
            rec = self._processed.get(idempotency_key)
            if rec is not None and not rec.done:
                del self._processed[idempotency_key]

    async def check_processed(self, idempotency_key: str) -> Optional[Any]:
        async with self._lock:
            rec = self._processed.get(idempotency_key)
            return rec.result if rec is not None and rec.done else None

    async def mark_processed(self, idempotency_key: str, result: Any) -> None:
        async with self._lock:
            rec = self._processed.get(idempotency_key)
            if rec is not None and rec.done:
                return  # first result wins
            self._processed[idempotency_key] = _Record(done=True, result=result)

StoragePlugin

Bases: Protocol

Unified interface for storage backends. Every storage must support all operations — the handler's durability parameter decides which ones are actually called.

Source code in src/istos/consistency/storage.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@runtime_checkable
class StoragePlugin(Protocol):
    """
    Unified interface for storage backends.
    Every storage must support all operations — the handler's `durability`
    parameter decides which ones are actually called.
    """

    # ---- Core key-value (always used) ----

    async def put(self, key: str, value: Any) -> None:
        """Write or overwrite a value by key."""
        ...

    async def get(self, key: str) -> Optional[Any]:
        """Retrieve a value by key."""
        ...

    async def delete(self, key: str) -> None:
        """Delete a key from storage."""
        ...

    # ---- Event log (used by AT_LEAST_ONCE and EXACTLY_ONCE) ----

    async def log(self, key: str, value: Any, idempotency_key: Optional[str] = None) -> None:
        """Append an event to the durable log. Skips duplicates by idempotency_key."""
        ...

    async def get_log(self, key: str, limit: int = 100) -> List[Any]:
        """Retrieve event log entries for a key, newest first."""
        ...

    # ---- Idempotency (used by EXACTLY_ONCE) ----

    async def claim_processed(
        self, idempotency_key: str, *, lease_s: float = DEFAULT_CLAIM_LEASE_S
    ) -> Claim:
        """Atomically claim ``idempotency_key`` for execution, or report its state.

        Decides in one indivisible step whether this caller is the one that
        executes. :meth:`check_processed` before the handler plus
        :meth:`mark_processed` after it is not equivalent: every concurrent
        duplicate passes the check and executes.

        The claim carries a lease, so a process that dies mid-handler does not
        block the key forever. DONE is terminal — a lease never expires a
        finished result.
        """
        ...

    async def release_claim(self, idempotency_key: str) -> None:
        """Give up a claim that did not finish, so the work can be retried.

        A no-op if the key is already DONE — a completed result is never undone.
        """
        ...

    async def check_processed(self, idempotency_key: str) -> Optional[Any]:
        """Cached result if the work *finished*, else None.

        A key that is only claimed reads as None. This cannot tell "never ran"
        from "ran and returned None" — use :meth:`claim_processed` when that
        difference matters.
        """
        ...

    async def mark_processed(self, idempotency_key: str, result: Any) -> None:
        """Complete a claim: mark the key DONE and cache ``result``.

        First result wins — a key that is already DONE is left untouched.
        """
        ...

check_processed(idempotency_key) async

Cached result if the work finished, else None.

A key that is only claimed reads as None. This cannot tell "never ran" from "ran and returned None" — use :meth:claim_processed when that difference matters.

Source code in src/istos/consistency/storage.py 
106
107
108
109
110
111
112
113
async def check_processed(self, idempotency_key: str) -> Optional[Any]:
    """Cached result if the work *finished*, else None.

    A key that is only claimed reads as None. This cannot tell "never ran"
    from "ran and returned None" — use :meth:`claim_processed` when that
    difference matters.
    """
    ...

claim_processed(idempotency_key, *, lease_s=DEFAULT_CLAIM_LEASE_S) async

Atomically claim idempotency_key for execution, or report its state.

Decides in one indivisible step whether this caller is the one that executes. :meth:check_processed before the handler plus :meth:mark_processed after it is not equivalent: every concurrent duplicate passes the check and executes.

The claim carries a lease, so a process that dies mid-handler does not block the key forever. DONE is terminal — a lease never expires a finished result.

Source code in src/istos/consistency/storage.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
async def claim_processed(
    self, idempotency_key: str, *, lease_s: float = DEFAULT_CLAIM_LEASE_S
) -> Claim:
    """Atomically claim ``idempotency_key`` for execution, or report its state.

    Decides in one indivisible step whether this caller is the one that
    executes. :meth:`check_processed` before the handler plus
    :meth:`mark_processed` after it is not equivalent: every concurrent
    duplicate passes the check and executes.

    The claim carries a lease, so a process that dies mid-handler does not
    block the key forever. DONE is terminal — a lease never expires a
    finished result.
    """
    ...

delete(key) async

Delete a key from storage.

Source code in src/istos/consistency/storage.py 
67
68
69
async def delete(self, key: str) -> None:
    """Delete a key from storage."""
    ...

get(key) async

Retrieve a value by key.

Source code in src/istos/consistency/storage.py 
63
64
65
async def get(self, key: str) -> Optional[Any]:
    """Retrieve a value by key."""
    ...

get_log(key, limit=100) async

Retrieve event log entries for a key, newest first.

Source code in src/istos/consistency/storage.py 
77
78
79
async def get_log(self, key: str, limit: int = 100) -> List[Any]:
    """Retrieve event log entries for a key, newest first."""
    ...

log(key, value, idempotency_key=None) async

Append an event to the durable log. Skips duplicates by idempotency_key.

Source code in src/istos/consistency/storage.py 
73
74
75
async def log(self, key: str, value: Any, idempotency_key: Optional[str] = None) -> None:
    """Append an event to the durable log. Skips duplicates by idempotency_key."""
    ...

mark_processed(idempotency_key, result) async

Complete a claim: mark the key DONE and cache result.

First result wins — a key that is already DONE is left untouched.

Source code in src/istos/consistency/storage.py 
115
116
117
118
119
120
async def mark_processed(self, idempotency_key: str, result: Any) -> None:
    """Complete a claim: mark the key DONE and cache ``result``.

    First result wins — a key that is already DONE is left untouched.
    """
    ...

put(key, value) async

Write or overwrite a value by key.

Source code in src/istos/consistency/storage.py 
59
60
61
async def put(self, key: str, value: Any) -> None:
    """Write or overwrite a value by key."""
    ...

release_claim(idempotency_key) async

Give up a claim that did not finish, so the work can be retried.

A no-op if the key is already DONE — a completed result is never undone.

Source code in src/istos/consistency/storage.py 
 99
100
101
102
103
104
async def release_claim(self, idempotency_key: str) -> None:
    """Give up a claim that did not finish, so the work can be retried.

    A no-op if the key is already DONE — a completed result is never undone.
    """
    ...