Graph Algorithms and Traversal Questions

import time
from collections import deque

def enumerate_paths(adj, src, dst, k, max_paths=None, timeout=None):
    """
    Enumerate simple directed paths from src to dst with at most k edges.
    adj: dict[node] -> iterable(neighbor)
    Returns a generator yielding lists of nodes.
    """
    start_time = time.time()
    seen_paths = 0
    stack = deque()
    stack.append((src, [src]))  # (current_node, path)

    while stack:
        if timeout and (time.time() - start_time) > timeout:
            return  # stop early on timeout
        node, path = stack.pop()
        if len(path)-1 > k:
            continue
        if node == dst:
            yield path
            seen_paths += 1
            if max_paths and seen_paths >= max_paths:
                return
            continue
        # expand neighbors (reverse to preserve natural order with stack)
        for nei in reversed(list(adj.get(node, []))):
            if nei in path:  # enforce simple path
                continue
            if len(path) < k+1:  # only add if adding respects k edges
                stack.append((nei, path + [nei]))

# config
DAMPING=0.85
EPS_NODE=1e-6       # local convergence threshold per node
MAX_HOPS=20         # propagation bounding (TTL)
RECONCILE_PERIOD=3600  # seconds for batch recompute

def on_edge_update(src, dst, op):
    # op = 'insert' or 'delete'
    # update out_deg atomically for source
    old_out = get_out_deg(src)
    new_out = old_out + (1 if op=='insert' else -1)
    set_out_deg(src, new_out)
    # compute delta effect on src's outgoing contribution
    # change in contribution = PR(src) * (1/old_out - 1/new_out)
    pr_src = get_PR(src)
    if old_out==0 and new_out==0:
        return
    old_contrib = pr_src / old_out if old_out>0 else 0
    new_contrib = pr_src / new_out if new_out>0 else 0
    delta = new_contrib - old_contrib
    # schedule delta to affected neighbor(s)
    schedule_delta(dst, delta, ttl=MAX_HOPS)

def schedule_delta(node, delta, ttl):
    if abs(delta) < EPS_NODE: return
    # accumulate pending delta and mark active
    acc = atomic_add_pending(node, delta)
    if abs(acc) >= EPS_NODE:
        push_to_worker_queue(node, ttl)

# Worker: consumes active nodes, applies pending delta and propagates
def worker_process(node, ttl):
    pending = atomic_swap_pending(node, 0)
    if abs(pending) < EPS_NODE: return
    pr = get_PR(node)
    # update PR using push-style delta (personalization ignored)
    new_pr = pr + pending
    set_PR(node, new_pr)
    set_last_update(node, now())
    if ttl<=0: return
    out_d = get_out_deg(node)
    if out_d==0: return
    # propagate portion to each neighbor: contribution = pending * DAMPING / out_d
    contrib = pending * DAMPING / out_d
    for nbr in get_out_neighbors(node):
        schedule_delta(nbr, contrib, ttl-1)

# Periodic bounding & local convergence:
# - limit per-node work rate by a token bucket or max updates per window
# - stop propagating if contribution magnitude < EPS_NODE or hop exceeded
# - use active_flag TTL to avoid reprocessing hot nodes endlessly

# Periodic reconciliation job (batch):
def reconcile_batch():
    # run full PageRank on current snapshot (e.g., Spark GraphX)
    full_PR = run_batch_pagerank_snapshot()
    for node, pr in full_PR.items():
        set_PR(node, pr)
        clear_pending(node)
        set_last_reconcile(node, now())

WITH RECURSIVE reachable(node, path) AS (
  -- base case: start node; path stores visited nodes as array
  SELECT src AS node, ARRAY[src]::int[] AS path
  FROM (VALUES (42)) AS start(src)  -- replace 42 with your start node_id

  UNION

  -- recursive step: follow edges from current node to neighbors not yet visited
  SELECT e.dst AS node, r.path || e.dst
  FROM reachable r
  JOIN edges e ON e.src = r.node
  WHERE NOT e.dst = ANY(r.path)
)
SELECT DISTINCT node
FROM reachable;