Skip to content

Commit be9bd0b

Browse files
polazpolaz
committed
feat(demo): notebook for what 0.5 added, and correct the time-travel docs
Covers the client surface 0.5 brought: create_nodes_batch, element_id, schema_revision, the write-concern ladder with measured latency, read concerns and read preference, and time travel. Every cell runs against the demo stack; the time-travel cell asserts the later write really is invisible in the past rather than printing something that looks right. Getting that cell to work turned up two things the docstring had wrong. at_timestamp is microseconds since the epoch, not the documented milliseconds-with-a-counter: a millisecond value silently reads the far past, and a value scaled the documented way reads the present and looks like a no-op. It also requires read_concern="snapshot"; anything else is refused with FAILED_PRECONDITION. Both are now in the docstring and called out in the notebook, since both cost an afternoon to find from the outside.
1 parent 5eb5d94 commit be9bd0b

2 files changed

Lines changed: 383 additions & 4 deletions

File tree

‎coordinode/coordinode/client.py‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -289,10 +289,12 @@ async def cypher(
289289
``"secondary_preferred"``, ``"nearest"``.
290290
- ``after_index``: raft log index for causal reads, a fence. Returned rows reflect at
291291
least the state at this index.
292-
- ``at_timestamp``: HLC timestamp to read at, a pin rather than a fence. Reads the
293-
database exactly as of that version without waiting, for time travel. The timestamp
294-
has to fall inside the MVCC retention window; older snapshots are collected and the
295-
server answers UNAVAILABLE.
292+
- ``at_timestamp``: timestamp to read at, a pin rather than a fence. Reads the
293+
database exactly as of that version without waiting, for time travel. Microseconds
294+
since the Unix epoch, so ``int(time.time() * 1_000_000)`` is now. Requires
295+
``read_concern="snapshot"``; any other level is rejected with FAILED_PRECONDITION.
296+
The timestamp has to fall inside the MVCC retention window; older snapshots are
297+
collected and the server answers UNAVAILABLE.
296298
"""
297299
from coordinode._proto.coordinode.v1.query.cypher_pb2 import ( # type: ignore[import]
298300
ExecuteCypherRequest,

‎demo/notebooks/04_whats_new_in_0_5.ipynb‎

Lines changed: 377 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,377 @@
1+
{
2+
"cells": [
3+
{
4+
"cell_type": "markdown",
5+
"metadata": {},
6+
"source": [
7+
"# What 0.5 Added\n",
8+
"\n",
9+
"[![Open in Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/structured-world/coordinode-python/blob/main/demo/notebooks/04_whats_new_in_0_5.ipynb)\n",
10+
"\n",
11+
"The client surface that arrived with CoordiNode 0.5, exercised end to end:\n",
12+
"\n",
13+
"| Feature | What it is for |\n",
14+
"|---------|----------------|\n",
15+
"| `create_nodes_batch` | Create many nodes in one atomic write instead of a loop of round trips |\n",
16+
"| `element_id` | A stable identifier that survives restarts and replication |\n",
17+
"| `schema_revision` | Tells you when a label's shape last changed, so caches know to refresh |\n",
18+
"| `write_concern` | Chooses how durable an acknowledgement is, from fire-and-forget to majority |\n",
19+
"| `read_concern` / `read_preference` | Chooses how fresh a read is, and which replica answers it |\n",
20+
"| `at_timestamp` | Reads the database as it was at a point in time |\n",
21+
"\n",
22+
"> **Needs a server.** These are distribution and durability features, so they\n",
23+
"> exist on the client that talks to a CoordiNode server. Set `COORDINODE_ADDR`\n",
24+
"> before running. The embedded engine has no Raft and no replicas, so the\n",
25+
"> cells below stop with an explanation rather than pretending.\n"
26+
]
27+
},
28+
{
29+
"cell_type": "markdown",
30+
"metadata": {},
31+
"source": [
32+
"## Install dependencies\n"
33+
]
34+
},
35+
{
36+
"cell_type": "code",
37+
"execution_count": null,
38+
"metadata": {},
39+
"outputs": [],
40+
"source": [
41+
"import importlib.util, os, subprocess, sys\n",
42+
"\n",
43+
"# (distribution on PyPI, module to import it by).\n",
44+
"pkgs = [\n",
45+
" (\"coordinode\", \"coordinode\"),\n",
46+
" (\"nest_asyncio\", \"nest_asyncio\"),\n",
47+
"]\n",
48+
"\n",
49+
"# Install only what is missing. A checkout mounted into the image, or an\n",
50+
"# editable install, already provides these; pulling them from PyPI there\n",
51+
"# would shadow the very code the notebook is meant to exercise.\n",
52+
"missing = [dist for dist, mod in pkgs if importlib.util.find_spec(mod) is None]\n",
53+
"if missing:\n",
54+
" subprocess.run(\n",
55+
" [sys.executable, \"-m\", \"pip\", \"install\", \"-q\", *missing],\n",
56+
" check=True,\n",
57+
" timeout=300,\n",
58+
" )\n",
59+
"\n",
60+
"import nest_asyncio\n",
61+
"\n",
62+
"nest_asyncio.apply()\n",
63+
"\n",
64+
"print(\"Ready\")\n"
65+
]
66+
},
67+
{
68+
"cell_type": "markdown",
69+
"metadata": {},
70+
"source": [
71+
"## Connect\n",
72+
"\n",
73+
"Every cell below guards on `client`, so running the notebook without a server\n",
74+
"produces one explanation instead of a cascade of failures.\n"
75+
]
76+
},
77+
{
78+
"cell_type": "code",
79+
"execution_count": null,
80+
"metadata": {},
81+
"outputs": [],
82+
"source": [
83+
"import os, time, uuid\n",
84+
"\n",
85+
"COORDINODE_ADDR = os.environ.get(\"COORDINODE_ADDR\")\n",
86+
"client = None\n",
87+
"\n",
88+
"if COORDINODE_ADDR:\n",
89+
" from coordinode import CoordinodeClient\n",
90+
"\n",
91+
" client = CoordinodeClient(COORDINODE_ADDR)\n",
92+
" if not client.health():\n",
93+
" client.close()\n",
94+
" raise RuntimeError(f\"Health check failed for {COORDINODE_ADDR}\")\n",
95+
" print(f\"Connected to {COORDINODE_ADDR}\")\n",
96+
"else:\n",
97+
" print(\n",
98+
" \"COORDINODE_ADDR is not set, so this notebook has nothing to talk to.\\n\"\n",
99+
" \"Start the demo stack (see demo/README.md) or point at your own server:\\n\"\n",
100+
" ' os.environ[\"COORDINODE_ADDR\"] = \"localhost:37080\"'\n",
101+
" )\n",
102+
"\n",
103+
"# Tag every node this run creates, so the cleanup at the end removes exactly\n",
104+
"# what was added and nothing a sibling notebook left behind.\n",
105+
"DEMO_TAG = f\"whats_new_{uuid.uuid4().hex[:8]}\"\n",
106+
"print(f\"DEMO_TAG: {DEMO_TAG}\")\n"
107+
]
108+
},
109+
{
110+
"cell_type": "markdown",
111+
"metadata": {},
112+
"source": [
113+
"## Atomic bulk insert\n",
114+
"\n",
115+
"`create_nodes_batch` sends the whole set as one write. A loop over\n",
116+
"`create_node` costs a round trip each and, more importantly, can leave half\n",
117+
"the batch committed if the process dies midway; the batch either lands\n",
118+
"completely or not at all.\n"
119+
]
120+
},
121+
{
122+
"cell_type": "code",
123+
"execution_count": null,
124+
"metadata": {},
125+
"outputs": [],
126+
"source": [
127+
"if client:\n",
128+
" people = [\n",
129+
" ([\"Engineer\"], {\"name\": name, \"team\": team, \"tag\": DEMO_TAG})\n",
130+
" for name, team in [\n",
131+
" (\"Ada\", \"storage\"),\n",
132+
" (\"Grace\", \"storage\"),\n",
133+
" (\"Linus\", \"kernel\"),\n",
134+
" (\"Barbara\", \"query\"),\n",
135+
" (\"Edsger\", \"query\"),\n",
136+
" ]\n",
137+
" ]\n",
138+
" created = client.create_nodes_batch(people)\n",
139+
" print(f\"Created {len(created)} nodes in one write\")\n",
140+
" for n in created[:3]:\n",
141+
" print(f\" {n.properties['name']:<8} id={n.id} element_id={n.element_id}\")\n"
142+
]
143+
},
144+
{
145+
"cell_type": "markdown",
146+
"metadata": {},
147+
"source": [
148+
"## Stable identity: `element_id`\n",
149+
"\n",
150+
"`id` is a numeric handle kept for Neo4j v4 driver compatibility. `element_id`\n",
151+
"is the canonical one: stable across restarts, schema changes and replication,\n",
152+
"and it is what application code should store when it needs to point at a node\n",
153+
"later. On an edge it names the two endpoints in canonical order, because an\n",
154+
"edge here is a typed property bag between two nodes rather than an entity with\n",
155+
"its own identity.\n"
156+
]
157+
},
158+
{
159+
"cell_type": "code",
160+
"execution_count": null,
161+
"metadata": {},
162+
"outputs": [],
163+
"source": [
164+
"if client:\n",
165+
" ada, grace = created[0], created[1]\n",
166+
" edge = client.create_edge(\"PAIRS_WITH\", ada.id, grace.id, {\"since\": 2026})\n",
167+
" print(f\"node element_id: {ada.element_id}\")\n",
168+
" print(f\"edge element_id: {edge.element_id} ({edge.type})\")\n",
169+
"\n",
170+
" # The handle survives a re-read: fetch the node again and compare.\n",
171+
" again = client.get_node(ada.id)\n",
172+
" print(f\"re-read matches: {again.element_id == ada.element_id}\")\n"
173+
]
174+
},
175+
{
176+
"cell_type": "markdown",
177+
"metadata": {},
178+
"source": [
179+
"## Schema revision\n",
180+
"\n",
181+
"Every label and edge type carries a revision that moves when its shape\n",
182+
"changes. A client that caches a schema compares revisions instead of\n",
183+
"re-fetching and re-parsing the whole thing.\n"
184+
]
185+
},
186+
{
187+
"cell_type": "code",
188+
"execution_count": null,
189+
"metadata": {},
190+
"outputs": [],
191+
"source": [
192+
"if client:\n",
193+
" labels = {l.name: l for l in client.get_labels()}\n",
194+
" engineer = labels.get(\"Engineer\")\n",
195+
" if engineer:\n",
196+
" print(f\"Engineer schema_revision: {engineer.schema_revision}\")\n",
197+
" for et in client.get_edge_types():\n",
198+
" if et.name == \"PAIRS_WITH\":\n",
199+
" print(f\"PAIRS_WITH schema_revision: {et.schema_revision}\")\n"
200+
]
201+
},
202+
{
203+
"cell_type": "markdown",
204+
"metadata": {},
205+
"source": [
206+
"## Write concerns: how durable is \"done\"\n",
207+
"\n",
208+
"A write concern is the answer to \"acknowledged by whom\". They rise in\n",
209+
"durability:\n",
210+
"\n",
211+
"| Concern | Acknowledged when |\n",
212+
"|---------|-------------------|\n",
213+
"| `w0` | The server accepted the request; nothing is guaranteed |\n",
214+
"| `memory` | Applied in memory, before Raft |\n",
215+
"| `cache` | Cached, still before Raft |\n",
216+
"| `w1` | The leader has it durably (default) |\n",
217+
"| `majority` | A majority of the cluster has it |\n",
218+
"\n",
219+
"`memory` and `cache` acknowledge before the write reaches Raft, so a leader\n",
220+
"crash before the background drain loses them. They are for data you can\n",
221+
"afford to lose, not for a speed-up on data you cannot.\n"
222+
]
223+
},
224+
{
225+
"cell_type": "code",
226+
"execution_count": null,
227+
"metadata": {},
228+
"outputs": [],
229+
"source": [
230+
"if client:\n",
231+
" for concern in (\"w0\", \"memory\", \"cache\", \"w1\", \"majority\"):\n",
232+
" started = time.perf_counter()\n",
233+
" client.cypher(\n",
234+
" \"CREATE (:Sample {tag: $tag, concern: $concern})\",\n",
235+
" params={\"tag\": DEMO_TAG, \"concern\": concern},\n",
236+
" write_concern=concern,\n",
237+
" )\n",
238+
" elapsed_ms = (time.perf_counter() - started) * 1000\n",
239+
" print(f\" {concern:<9} acknowledged in {elapsed_ms:6.2f} ms\")\n"
240+
]
241+
},
242+
{
243+
"cell_type": "markdown",
244+
"metadata": {},
245+
"source": [
246+
"## Read concerns and read preference\n",
247+
"\n",
248+
"A read concern says how fresh the answer must be; a read preference says which\n",
249+
"node may answer. On the single-node demo stack every combination returns the\n",
250+
"same rows, which is the point: the same code runs unchanged against a cluster,\n",
251+
"where the choice starts to matter.\n"
252+
]
253+
},
254+
{
255+
"cell_type": "code",
256+
"execution_count": null,
257+
"metadata": {},
258+
"outputs": [],
259+
"source": [
260+
"if client:\n",
261+
" for concern in (\"local\", \"majority\", \"linearizable\", \"snapshot\"):\n",
262+
" rows = client.cypher(\n",
263+
" \"MATCH (n:Sample {tag: $tag}) RETURN count(n) AS n\",\n",
264+
" params={\"tag\": DEMO_TAG},\n",
265+
" read_concern=concern,\n",
266+
" )\n",
267+
" print(f\" read_concern={concern:<13} -> {rows[0]['n']} rows\")\n",
268+
"\n",
269+
" rows = client.cypher(\n",
270+
" \"MATCH (n:Sample {tag: $tag}) RETURN count(n) AS n\",\n",
271+
" params={\"tag\": DEMO_TAG},\n",
272+
" read_preference=\"primary_preferred\",\n",
273+
" )\n",
274+
" print(f\" read_preference=primary_preferred -> {rows[0]['n']} rows\")\n"
275+
]
276+
},
277+
{
278+
"cell_type": "markdown",
279+
"metadata": {},
280+
"source": [
281+
"## Time travel: `at_timestamp`\n",
282+
"\n",
283+
"`at_timestamp` pins a read to a version of the database instead of waiting for\n",
284+
"one. Two things it is easy to get wrong:\n",
285+
"\n",
286+
"- The value is **microseconds since the Unix epoch**, so `int(time.time() * 1_000_000)`\n",
287+
" is now. Pass a millisecond value and the read silently lands in the far past;\n",
288+
" pass one scaled too high and it lands in the present and looks like a no-op.\n",
289+
"- It requires `read_concern=\"snapshot\"`. Any other level is refused with\n",
290+
" `FAILED_PRECONDITION`, because reading at a pinned version is exactly what a\n",
291+
" snapshot read is.\n",
292+
"\n",
293+
"The timestamp also has to fall inside the MVCC retention window. Ask for\n",
294+
"something older than the server still keeps and it answers `UNAVAILABLE` rather\n",
295+
"than quietly returning the oldest thing it has.\n"
296+
]
297+
},
298+
{
299+
"cell_type": "code",
300+
"execution_count": null,
301+
"metadata": {},
302+
"outputs": [],
303+
"source": [
304+
"def now_micros() -> int:\n",
305+
" \"\"\"The timestamp CoordiNode reads at: microseconds since the Unix epoch.\"\"\"\n",
306+
" return int(time.time() * 1_000_000)\n",
307+
"\n",
308+
"\n",
309+
"if client:\n",
310+
" client.cypher(\n",
311+
" \"CREATE (:Era {tag: $tag, name: 'before'})\",\n",
312+
" params={\"tag\": DEMO_TAG},\n",
313+
" write_concern=\"majority\",\n",
314+
" )\n",
315+
" time.sleep(1.5)\n",
316+
" mark = now_micros()\n",
317+
" time.sleep(1.5)\n",
318+
" client.cypher(\n",
319+
" \"CREATE (:Era {tag: $tag, name: 'after'})\",\n",
320+
" params={\"tag\": DEMO_TAG},\n",
321+
" write_concern=\"majority\",\n",
322+
" )\n",
323+
"\n",
324+
" query = \"MATCH (n:Era {tag: $tag}) RETURN n.name AS name ORDER BY name\"\n",
325+
" now = [r[\"name\"] for r in client.cypher(query, params={\"tag\": DEMO_TAG})]\n",
326+
" past = [\n",
327+
" r[\"name\"]\n",
328+
" for r in client.cypher(\n",
329+
" query,\n",
330+
" params={\"tag\": DEMO_TAG},\n",
331+
" at_timestamp=mark,\n",
332+
" read_concern=\"snapshot\",\n",
333+
" )\n",
334+
" ]\n",
335+
" print(f\" now : {now}\")\n",
336+
" print(f\" at the mark : {past}\")\n",
337+
" assert \"after\" not in past, \"the later write must be invisible in the past\"\n",
338+
" print(\" the write made after the mark is invisible there\")\n"
339+
]
340+
},
341+
{
342+
"cell_type": "markdown",
343+
"metadata": {},
344+
"source": [
345+
"## Clean up\n"
346+
]
347+
},
348+
{
349+
"cell_type": "code",
350+
"execution_count": null,
351+
"metadata": {},
352+
"outputs": [],
353+
"source": [
354+
"if client:\n",
355+
" for label in (\"Engineer\", \"Sample\", \"Era\"):\n",
356+
" client.cypher(\n",
357+
" f\"MATCH (n:{label} {{tag: $tag}}) DETACH DELETE n\",\n",
358+
" params={\"tag\": DEMO_TAG},\n",
359+
" )\n",
360+
" client.close()\n",
361+
" print(f\"Removed everything tagged {DEMO_TAG}\")\n"
362+
]
363+
}
364+
],
365+
"metadata": {
366+
"kernelspec": {
367+
"display_name": "Python 3",
368+
"language": "python",
369+
"name": "python3"
370+
},
371+
"language_info": {
372+
"name": "python"
373+
}
374+
},
375+
"nbformat": 4,
376+
"nbformat_minor": 5
377+
}

0 commit comments

Comments
 (0)