Skip to content

symfonic.memory.backends.postgres_linked_ack

postgres_linked_ack

Revalidate a linked source, published edge, and target inside acceptance.

linked_line_on async 

linked_line_on(conn, scope, publication, item, port)

Return the exact linked bytes from the locked source, edge and target.

Source code in src/symfonic/memory/backends/postgres_linked_ack.py 
 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
121
122
123
124
async def linked_line_on(conn, scope, publication, item, port):
    """Return the exact linked bytes from the locked source, edge and target."""
    schema = port.schema
    namespace = quoted(schema)
    edge = item["edge"]
    source_ref, target_ref = _ref(edge["source_publication"]), _ref(edge["target_publication"])
    rows = {}
    for ref in sorted((source_ref, target_ref), key=lambda value: value.record_id):
        rows[ref.record_id] = await _node(conn, scope, ref, schema)
    stored = await conn.fetchrow(
        f'SELECT * FROM ONLY {namespace}.memory_edges '
        'WHERE tenant_id COLLATE "C"=$1 AND id COLLATE "C"=$2 FOR SHARE',
        scope.tenant, edge["id"],
    )
    if (stored is None or stored["scope_path"] != edge["owner"]
            or stored["publication_format"] != 1
            or stored["publication_incarnation"] != UUID(edge["incarnation"])
            or stored["publication_revision"] != edge["revision"]
            or stored["source"] != source_ref.record_id
            or stored["target"] != target_ref.record_id
            or stored["relationship"] != edge["relationship"]):
        _refuse()
    anchor = source_ref if edge["direction"] == "outgoing" else target_ref
    neighbor = target_ref if edge["direction"] == "outgoing" else source_ref
    if publication != neighbor:
        _refuse()
    target = rows[neighbor.record_id]
    record = record_from_legacy_node(node_payload(target))
    content = target.properties.get("content", record.text)
    if type(content) is not str or not content.strip():
        _refuse()
    record = replace(record, text=content)
    value = linked_line(SimpleNamespace(
        source_text=rows[anchor.record_id].label,
        relationship=edge["relationship"], record=record,
    )).encode("utf-8")
    if (hashlib.sha256(value).hexdigest() != item["line_digest"]
            or len(value) != item["end"] - item["start"]):
        _refuse()
    return value

prepare_linked_on async 

prepare_linked_on(conn, scope, records, port)

Attest one schema and lock every row before any access increment.

Source code in src/symfonic/memory/backends/postgres_linked_ack.py 
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
async def prepare_linked_on(conn, scope, records, port):
    """Attest one schema and lock every row before any access increment."""
    from symfonic.memory.backends.postgres_linked_edges_v1 import PostgresLinkedEdgesV1

    if type(port) is not PostgresLinkedEdgesV1 or not port.schema:
        _refuse()
    schema = port.schema
    if await conn.fetchval("SELECT current_schema()") != schema:
        _refuse()
    search_path = await conn.fetchval("SELECT current_setting('search_path')")
    namespace = quoted(schema)
    await conn.execute(f"LOCK TABLE {namespace}.memory_nodes, {namespace}.memory_edges "
                       "IN ROW EXCLUSIVE MODE")
    await attest(conn, schema, working_guard=port.working_guard)
    await conn.fetchval("SELECT set_config('search_path',$1,true)", search_path)
    ids = {publication.record_id for publication, _ in records}
    for _, item in records:
        if "edge" in item:
            ids.update((item["edge"]["source"], item["edge"]["target"]))
    await conn.fetch(
        f'SELECT id FROM ONLY {namespace}.memory_nodes '
        'WHERE tenant_id COLLATE "C"=$1 AND id=ANY($2::text[]) '
        'ORDER BY id COLLATE "C" FOR UPDATE',
        scope.tenant, sorted(ids),
    )