sql/pgq_node: doc: describe workflow for cascaded consumer

1 files changed, 29 insertions, 0 deletions

diff --git a/sql/pgq_node/structure/functions.sql b/sql/pgq_node/structure/functions.sql
index 6196d3d4..821a5656 100644
--- a/sql/pgq_node/structure/functions.sql
+++ b/sql/pgq_node/structure/functions.sql
@@ -1,6 +1,35 @@
 -- File: Functions
 --
 --      Database functions for cascaded pgq.
+--
+-- Cascaded consumer flow:
+--
+--  - (1) [target] call pgq_node.get_consumer_state()
+--  - (2) If .paused is true, sleep, go to (1).
+--    This is allows to control consumer remotely.
+--  - (3) If .uptodate is false, call pgq_node.set_consumer_uptodate(true).
+--    This allows remote controller to know that consumer has seen the changes.
+--  - (4) [source] call pgq.next_batch().  If returns NULL, sleep, goto (1)
+--  - (5) [source] if batch already done, call pgq.finish_batch(), go to (1)
+--  - (6) [source] read events
+--  - (7) [target] process events, call pgq_node.set_consumer_completed() in same tx.
+--  - (8) [source] call pgq.finish_batch()
+--
+-- Cascaded worker flow:
+--
+-- Worker is consumer that also copies to queue contents to local node (branch),
+-- so it can act as provider to other nodes.  There can be only one worker per
+-- node.  Or zero if node is leaf.  In addition to cascaded consumer logic above, it must -
+--      - [branch] copy all events to local queue and create ticks
+--      - [merge-leaf] copy all events to combined-queue
+--      - [branch] publish local watermark upwards to provider so it reaches root.
+--      - [branch] apply global watermark event to local node
+--      - [merge-leaf] wait-behind on combined-branch (failover combined-root).
+--        It's last_tick_id is set by combined-branch worker, it must call
+--        pgq.next_batch()+pgq.finish_batch() without processing events
+--        when behind, but not move further.  When the combined-branch
+--        becomes root, it will be in right position to continue updating.
+--
 
 \i functions/pgq_node.upgrade_schema.sql
 select pgq_node.upgrade_schema();