Gateway

For advanced deployments and contributors. Documents the Python sidecar that bridges Odoo JSON-RPC with the Kubo daemon.

---

File Structure

gateway/
├── main.py           # Entry point, asyncio orchestrator
├── api/
│   └── express.py    # FastAPI endpoints + Bouncer
├── network/
│   ├── ipfs.py       # Kubo RPC client (async)
│   └── odoo.py       # Odoo JSON-RPC client (dual sync+async)
├── workers/
│   ├── listener.py   # Gossipsub subscriber (async)
│   ├── verifier.py   # DHT size-check for unknown CIDs (async)
│   ├── fetcher.py    # CID downloader (async, priority queue)
│   └── batcher.py    # State sync with Odoo (sync thread, 5s loop)
└── common/
    ├── bouncer.py    # Unified inbound firewall
    └── state.py      # Thread-safe in-memory topic+peer caches

---

Event & CID Routing

Outbound (Odoo → Network)

           ┌── Batcher (polling) ──────┐
ip.event ──┤                           ├── bcast_queue ───▶ Broadcaster ─▶ PubSub
           └── Express (webhook) ──────┤
                                       └── direct_queue ──▶ Sender ──────▶ P2P


fields.Binary (Odoo) ──▶ add() ──▶ ip.cid (pinned)
fields.Binary (Odoo) ──▶ Express ──▶ fetch_queue ──▶ Fetcher ──▶ cat() ──▶ ip.cid (pinned)

Inbound (Network → Odoo)

PubSub ──▶ Listener ────────────▶ Bouncer ──▶ batch_queue ──▶ Batcher ────────┐
                                                                              ├─▶ ip.event
P2P ──▶ Express (p2p_receive) ──▶ Bouncer ──▶ Express (route_inbound_event) ──┘


CID ───────────▶ verif_queue ──▶ Verifier ──┐
(unknown peer)       ┌──────────────────────┘
                     ▼
CID ───────────▶ fetch_queue ──▶ Fetcher ──▶ batch_queue ──▶ Batcher ──▶ ip.cid (pinned)
(trusted peer)

---

Queues

Queue	Content	Filled by	Consumed by
sync_batch_queue queue.Queue (thread-safe)	Inbound events, CID state updates, and node state changes crossing the async→sync boundary into the Batcher thread.	Listener, Express API, Fetcher, Verifier	Batcher
async_bcast_queue asyncio.Queue	Outbound broadcast events to be published on IPFS PubSub.	Batcher, Express API	Broadcaster
async_direct_queue asyncio.Queue	Outbound direct events to be sent to a specific peer over a Kubo P2P TCP tunnel.	Batcher, Express API	Sender
async_verif_queue asyncio.Queue	CID task dicts from unknown peers awaiting a DHT size check before being promoted to the fetch queue.	Listener	Verifier
async_fetch_queue asyncio.PriorityQueue	CID task dicts ready for download, with priority 0 for express/P2P CIDs and priority 1 for all others.	Verifier, Express API, Batcher	Fetcher

sync_batch_queue is the only thread-safe queue; it bridges the async worker coroutines with the synchronous OS-thread Batcher. All other queues are asyncio-native and live entirely within the event loop.

---

Boot Sequence

async_main()

Set up the async queues and boot all async task pools.

main()

Initiate the hybrid sync/async boot sequence.

---

Common

class OdooClient

Synchronous client for Odoo’s /json/2 API. Designed to run inside a dedicated background thread (The Batcher).

Methods

sync(sync_payload: dict, last_sync: datetime, timeout: float = 320) → dict

Call the custom /sync controller.

check_odoo_errors(data)

async_post(endpoint_path: str, params: dict, timeout: float = 5.0) → dict

Non-blocking POST request for the Express FastAPI layer.

class IpfsClient

Asynchronous client for the local Kubo RPC API. Handles PubSub streaming, DHT lookups, and Swarm connection management.

Core API

cat(cid: str, timeout: int = 30) → bytes

Download a CID payload from the IPFS network. Returns the raw file bytes.

stat(cid: str) → int

Check the total size of a CID on the network without downloading it. Returns the size in bytes.

PubSub

subscribe(topic: str)

Subscribe to a Gossipsub topic. Yields events as an async generator.

publish(topic: str, payload: bytes)

P2P

send_direct(to_peer: str, payload: bytes) → bool

Tunnels an HTTP POST request to a peer using the universal protocol.

Swarm Control

ban_peer(peer_id: str)

Identify peer’s IP, add it to the firewall, and disconnect.

unban_peer(peer_id: str)

Remove peer’s IP from Kubo’s network filter.

Lifecycle

wait_for_ipfs()

Asynchronously polls Kubo until it is ready to accept commands.

setup_p2p_listener()

Bind the gateway container to Kubo’s multiplexer.

class TopicState

Attributes

_active_topics: set

Names of currently subscribed PubSub topics (internal; use the topics property for thread-safe read access).

property topics

Thread-safe read of _active_topics; returns a frozenset.

Methods

is_active(topic: str) → bool

sync_from_odoo(new_topics: set)

Atomically replace the active topic set from an Odoo pull response.

Parameters:

new_topics – Iterable of dicts with name and active keys as returned by the batcher’s pull payload.

class PeerState

Attributes

trusted_peers: set

PeerIDs that bypass the verifier queue (fast lane).

blocked_peers: set

PeerIDs whose payloads are dropped at layer 2.

last_sync: bool

Watermark timestamp of the last successful batcher sync cycle.

peer_self: bool

Own node’s PeerID; used by the listener echo-catcher to identify messages broadcast by this gateway.

Methods

is_blocked(peer_id: str) → bool

is_trusted(peer_id: str) → bool

sync_from_odoo(new_nodes: set)

Atomically update trusted/blocked peer sets from an Odoo pull response.

Parameters:

new_nodes – Iterable of dicts with peer_char and trust_level keys as returned by the batcher’s pull payload.

class UnifiedBouncer

The Gateway Firewall. Protects FastAPI and Odoo from malformed payloads, oversized requests, and P2P DDoS attacks.

Attributes

max_direct_payload_bytes: int = 2097152

Maximum payload size in bytes for the express (direct) lane (2 MB).

max_requests_per_window: int = 10

Maximum requests per peer per rate_limit_window second(s).

Methods

admit(envelope: dict, routing: str, content_length: int = 0) → tuple[bool, str, dict, list]

Evaluate an incoming event envelope through the 5-layer firewall.

Layers are evaluated in order; the first failure short-circuits and the envelope is dropped:

1. Transport — direct lane only: content-length check against max_direct_payload_bytes (2 MB) and token-bucket rate limit (max_requests_per_window per second) per peer.

2. Identity — peer ID looked up in blocked_peers (O(1) in-memory set); blocked peers are dropped before any further processing.

3. Topic — decoded topic name verified against topics; packets for topics this gateway is not subscribed to are discarded.

4. Payload — base64-decode → JSON parse; malformed payloads are dropped.

5. CID extraction — regex scan for CID strings (Qm… / bafy…) in the decoded JSON; O(1) and schema-agnostic.

Semantic validation (Ed25519 signature, JSON Schema) is delegated to Odoo’s application layer inside PostgreSQL savepoints.

Parameters:

• envelope – The PubSub or P2P envelope dict.

• routing – 'direct' or 'broadcast' — controls which layers apply.

• content_length – Raw request byte size; checked against max_direct_payload_bytes on the direct lane.

Returns:

4-tuple (is_admitted, rejection_reason, odoo_event, cid_list) where is_admitted is True when the event passes all layers.

---

Workers

Batcher

sync_batcher_thread(odoo: OdooClient, sync_batch_queue: queue.Queue, async_bcast_queue: asyncio.Queue, async_direct_queue: asyncio.Queue, async_fetch_queue: asyncio.PriorityQueue, loop: asyncio.AbstractEventLoop, interval: float = 5.0)

Run synchronous Odoo Batcher (OS thread).

Runs a push/pull loop on a fixed interval (default 5 s):

1. Drain — pull everything out of sync_batch_queue into local buffers (inbound events, CID state updates, node state changes).

2. Push — send the accumulated batch to Odoo in a single JSON-RPC POST to /odoo_sync/v0/gateway. Calls ip.event._sync_push() (upsert on (from_peer, seqno)) and ip.cid state updates, each inside its own savepoint so a failure in one model never rolls back the others.

3. Pull — read the response and apply it locally: update state.topics and state.peer_state, enqueue wanted CIDs into async_fetch_queue, and route outbound events to async_bcast_queue (broadcast) or async_direct_queue (direct).

Parameters:

• odoo – Authenticated Odoo JSON-RPC client.

• sync_batch_queue – Thread-safe queue shared with async workers; carries inbound events, CID updates, and state change notifications.

• async_bcast_queue – Asyncio queue for broadcast events to publish.

• async_direct_queue – Asyncio queue for direct (express) events.

• async_fetch_queue – Priority queue for CIDs to download.

• loop – The running asyncio event loop (used to enqueue work thread-safely from this synchronous thread).

• interval – Seconds between sync cycles (default 5.0).

_log_odoo_response(odoo_status: dict)

Log the results of Odoo’s atomic savepoints.

Listener

async_listener_pool(ipfs: IpfsClient, bouncer: UnifiedBouncer, async_fetch_queue: asyncio.PriorityQueue, async_verif_queue: asyncio.Queue, sync_batch_queue: queue.Queue)

Manage listener pool — subscribe to every active PubSub topic.

Maintains a one-task-per-topic set of _single_listener_thread() coroutines. Polls topic_state every 5 s; spawns tasks for new topics and cancels tasks for topics that are no longer active.

Parameters:

• ipfs – Async Kubo RPC client.

• bouncer – Unified firewall shared across all listener tasks.

• async_fetch_queue – Priority queue for CIDs ready to download.

• async_verif_queue – Queue for CIDs pending a DHT size check.

• sync_batch_queue – Thread-safe queue to push events to the batcher.

process_listener_message(ipfs: IpfsClient, bouncer: UnifiedBouncer, topic: str, message: dict, async_fetch_queue: asyncio.PriorityQueue, async_verif_queue: asyncio.Queue, sync_batch_queue: queue.Queue)

Dispatch an incoming PubSub message to the appropriate processing lane.

The bouncer first admits or discards the message. Admitted messages are routed by trust level:

• Trusted peer → straight to async_fetch_queue (fast lane).

• Unknown peer → async_verif_queue for a DHT size-check first (slow lane).

Parameters:

• ipfs – Async Kubo RPC client used for stat/cat operations.

• bouncer – Unified firewall that validates the envelope.

• topic – The PubSub topic string the message arrived on.

• message – Raw PubSub message dict from Kubo.

• async_fetch_queue – Priority queue for CIDs ready to download.

• async_verif_queue – Queue for CIDs pending a DHT size check.

• sync_batch_queue – Thread-safe queue to push admitted events to Odoo.

_single_listener_thread(ipfs: IpfsClient, bouncer: UnifiedBouncer, topic: str, async_fetch_queue: asyncio.PriorityQueue, async_verif_queue: asyncio.Queue, sync_batch_queue: queue.Queue)

Dedicated infinite async loop for a single PubSub topic stream.

Subscribes to topic via ipfs.subscribe() and passes every incoming message through process_listener_message(). Reconnects automatically after network errors; exits cleanly on CancelledError.

Parameters:

• ipfs – Async Kubo RPC client.

• bouncer – Unified firewall shared across all listener tasks.

• topic – The PubSub topic path to subscribe to.

• async_fetch_queue – Priority queue for CIDs ready to download.

• async_verif_queue – Queue for CIDs pending a DHT size check.

• sync_batch_queue – Thread-safe queue to push events to the batcher.

Verifier

async_verifier_pool(ipfs: IpfsClient, async_verif_queue: asyncio.Queue, async_fetch_queue: asyncio.PriorityQueue, sync_batch_queue, max_tasks: int = 3)

Manage verifier pool — DHT-stat CIDs from unknown peers before fetching.

Spawns max_tasks concurrent _single_verifier_task() coroutines.

Parameters:

• ipfs – Async Kubo RPC client.

• async_verif_queue – Queue of CID tasks waiting for a size check.

• async_fetch_queue – Priority queue to promote verified CIDs into.

• sync_batch_queue – Thread-safe queue to push state updates to Odoo.

• max_tasks – Number of concurrent verifier coroutines (default 3).

_single_verifier_task(ipfs: IpfsClient, async_verif_queue: asyncio.Queue, async_fetch_queue: asyncio.PriorityQueue, sync_batch_queue)

Single verifier coroutine — DHT-stats each unknown CID before fetching.

Pulls tasks from async_verif_queue, calls ipfs.stat() to get the blob size, marks oversized CIDs (> 100 MB) as oversized, and promotes acceptable CIDs to async_fetch_queue at priority 1 (below trusted-peer priority 0).

Parameters:

• ipfs – Async Kubo RPC client.

• async_verif_queue – Queue of CID tasks from unknown peers.

• async_fetch_queue – Priority queue to promote verified CIDs into.

• sync_batch_queue – Thread-safe queue to push oversized / failed state updates to the batcher.

Fetcher

async_fetcher_pool(ipfs: IpfsClient, async_fetch_queue: asyncio.PriorityQueue, sync_batch_queue, max_tasks: int = 5)

Fetcher pool manager — concurrently downloads CIDs from the IPFS network.

Spawns max_tasks concurrent _single_fetcher_task() coroutines and keeps them running with asyncio.gather().

Parameters:

• ipfs – Async Kubo RPC client.

• async_fetch_queue – Priority queue of (rank, timestamp, tie) / task-dict tuples to download.

• sync_batch_queue – Thread-safe queue to push ip.cid state updates (pinned / failed / oversized) to the batcher.

• max_tasks – Number of concurrent downloader coroutines (default 5).

process_fetch_task(ipfs: IpfsClient, queue_event: tuple, async_fetch_queue: asyncio.PriorityQueue, sync_batch_queue)

Download one CID and report the result to the batcher.

Unpacks the priority-queue tuple, stats the blob size, downloads the raw bytes via ipfs.cat(), base64-encodes them (to avoid Postgres null-byte crashes), and enqueues an update action for ip.cid. On network failure the task re-inserts itself into async_fetch_queue up to retry_max times before recording a permanent failed state.

Parameters:

• ipfs – Async Kubo RPC client.

• queue_event – The raw item dequeued from async_fetch_queue. Format: ((rank, timestamp, tie_breaking), task_dict).

• async_fetch_queue – The priority queue to re-insert on failure.

• sync_batch_queue – Thread-safe queue to push state updates to Odoo.

_single_fetcher_task(ipfs: IpfsClient, async_fetch_queue: asyncio.PriorityQueue, sync_batch_queue)

Dedicated infinite loop for a single async fetcher task.

Broadcaster

async_broadcaster_pool(ipfs: IpfsClient, async_bcast_queue: asyncio.Queue, sync_batch_queue: queue.Queue, max_tasks: int = 2)

Broadcaster pool manager — publishes outbound broadcast events to IPFS PubSub.

Spawns max_tasks concurrent _single_broadcaster_task() coroutines.

Parameters:

• ipfs – Async Kubo RPC client.

• async_bcast_queue – Queue of outbound broadcast event dicts.

• sync_batch_queue – Thread-safe queue to push echo-matched sent state updates to the Batcher.

• max_tasks – Number of concurrent publisher coroutines (default 2).

_single_broadcaster_task(ipfs: IpfsClient, async_bcast_queue: asyncio.Queue, sync_batch_queue: queue.Queue)

Single broadcaster coroutine — publishes one broadcast event to IPFS PubSub.

Awaits events from async_bcast_queue, records an echo-hash in pending_broadcasts so the Listener can match the returning echo, then calls ipfs.publish(). The event is only marked sent once the Listener catches the echo, not here.

Sender

async_sender_pool(ipfs, async_direct_queue: asyncio.Queue, sync_batch_queue, num_workers=5)

Sender pool manager — delivers outbound direct events over Kubo P2P streams.

Spawns num_workers concurrent worker coroutines, each of which awaits events from async_direct_queue, builds a mock PubSub envelope, and dials the target peer via ipfs.send_direct(). On success it writes routing_state = sent back to Odoo via sync_batch_queue.

Parameters:

• ipfs – Async Kubo RPC client.

• async_direct_queue – Queue of outbound direct event dicts.

• sync_batch_queue – Thread-safe queue to push delivery confirmations to the Batcher.

• num_workers – Number of concurrent dial coroutines (default 5).

---

Express API

The FastAPI application is the only component that writes to queues from outside the asyncio worker pool, bridging external callers (Odoo, remote Kubo peers) into the gateway pipeline.

Endpoint	Caller	Queue(s) written
express_fetch	Odoo (ip.node._gate_request)	async_fetch_queue at priority 0 — one entry per CID. Bypasses the verifier; used when a waiting ip.event already arrived and the CID is needed urgently.
express_send	Odoo (ip.swarm._dispatch_events)	async_bcast_queue for broadcast events; async_direct_queue for direct events. The batcher’s polling loop is the slow path; this webhook is the fast path.
p2p_receive	Remote Kubo peer (libp2p dial)	async_fetch_queue at priority 0 for each CID in the envelope, then sync_batch_queue (via route_inbound_event) if the direct Odoo push times out.
route_inbound_event	p2p_receive (background task)	sync_batch_queue — only on fallback when the direct /odoo_sync/v0/push call fails or times out.

See also

Gateway — create_express_app, express_fetch, express_send, p2p_receive, and route_inbound_event endpoint reference.