Skip to content

symfonic.capabilities.memory.journal_node_patch

journal_node_patch

Conditional node patches recorded against a durable read guard.

bind_graph 

bind_graph(graph: Any) -> None

Expose the optional port only on wrappers with a capable durable store.

Source code in src/symfonic/capabilities/memory/journal_node_patch.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def bind_graph(graph: Any) -> None:
    """Expose the optional port only on wrappers with a capable durable store."""
    async def read_owned_node(scope: Any, node_id: Any) -> Any:
        from symfonic.capabilities.memory.journal import current_journal
        journal = current_journal()
        if journal is None:
            return await graph.durable.read_owned_node(scope, node_id)
        return await journal.read_owned_node(graph.durable, scope, node_id)

    async def patch_owned_node(scope: Any, node_id: Any, expected: OwnedNodeVersion,
                               properties: dict[str, Any]) -> Any:
        from symfonic.capabilities.memory.journal import current_journal
        journal = current_journal()
        if journal is None:
            return await graph.durable.patch_owned_node(scope, node_id, expected, properties)
        return await journal.patch_owned_node(
            graph.durable, scope, node_id, expected, properties,
        )

    graph.read_owned_node = read_owned_node
    graph.patch_owned_node = patch_owned_node

patch async 

patch(journal: Any, durable: Any, scope: Any, node_id: Any, expected: OwnedNodeVersion, properties: dict[str, Any]) -> Any

Stage property changes; replay uses the original durable version.

Source code in src/symfonic/capabilities/memory/journal_node_patch.py 
 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
async def patch(journal: Any, durable: Any, scope: Any, node_id: Any,
                expected: OwnedNodeVersion, properties: dict[str, Any]) -> Any:
    """Stage property changes; replay uses the original durable version."""
    if type(expected) is not OwnedNodeVersion:
        raise TypeError("patch requires an owned node version")
    if not isinstance(properties, dict) or any(
        reserved in properties for reserved in (SCOPE_PATH_KEY, "__spreading_access_count")
    ):
        raise ValueError("patch requires user properties without reserved keys")
    captured = journal._node_versions.get(key(durable, scope, node_id))
    if captured is None or expected != captured:
        return None
    refuse_conflict(journal, durable, scope, node_id)
    overlay = await journal._overlay(durable, scope)
    if await overlay.get_node(scope, node_id) is None:
        return None
    projected = await overlay.read_owned_node(scope, node_id)
    if projected is None:
        return None
    _, overlay_version = projected
    proposal = deepcopy(properties)
    applied = await overlay.patch_owned_node(scope, node_id, overlay_version, proposal)
    if applied is None:
        return None
    journal._require_open()
    journal._ops.setdefault(id(durable), []).append(
        (durable, "patch_owned_node", (scope, node_id, captured, proposal), {})
    )
    return applied

read async 

read(journal: Any, durable: Any, scope: Any, node_id: Any) -> Any

Capture a detached row and immutable guard before overlay writes occur.

Source code in src/symfonic/capabilities/memory/journal_node_patch.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
async def read(journal: Any, durable: Any, scope: Any, node_id: Any) -> Any:
    """Capture a detached row and immutable guard before overlay writes occur."""
    location = key(durable, scope, node_id)
    version = journal._node_versions.get(location)
    if version is None:
        result = await durable.read_owned_node(scope, node_id)
        if result is None:
            return None
        node, version = result
        if type(version) is not OwnedNodeVersion:
            raise TypeError("owned node read requires an OwnedNodeVersion")
        journal._node_versions[location] = version
    overlay = await journal._overlay(durable, scope)
    projected = await overlay.get_node(scope, node_id)
    return None if projected is None else (deepcopy(projected), version)

refuse_after_patch 

refuse_after_patch(journal: Any, durable: Any, name: str, args: tuple[Any, ...], kwargs: dict[str, Any]) -> None

Do not queue a same-row write after a guarded patch was promised.

Source code in src/symfonic/capabilities/memory/journal_node_patch.py 
62
63
64
65
66
67
68
69
70
def refuse_after_patch(journal: Any, durable: Any, name: str,
                       args: tuple[Any, ...], kwargs: dict[str, Any]) -> None:
    """Do not queue a same-row write after a guarded patch was promised."""
    target = _target(name, args, kwargs)
    if target is not None and any(
        previous == "patch_owned_node" and _target(previous, queued, queued_kw) == target
        for _, previous, queued, queued_kw in journal._ops.get(id(durable), ())
    ):
        raise MemoryContractError("conflicting staged mutation of the same node")

refuse_conflict 

refuse_conflict(journal: Any, durable: Any, scope: Any, node_id: Any) -> None

A guarded patch cannot replay against its own earlier node revision.

Source code in src/symfonic/capabilities/memory/journal_node_patch.py 
54
55
56
57
58
59
def refuse_conflict(journal: Any, durable: Any, scope: Any, node_id: Any) -> None:
    """A guarded patch cannot replay against its own earlier node revision."""
    target = scope.tenant_id, str(node_id)
    if any(_target(name, args, kwargs) == target
           for _, name, args, kwargs in journal._ops.get(id(durable), ())):
        raise MemoryContractError("conflicting staged mutation of the same node")