From 105bbdab4cfcfed6374124f60373e9fdced87de7 Mon Sep 17 00:00:00 2001 From: Pavan Kumar Date: Wed, 20 May 2026 12:37:37 +0530 Subject: [PATCH] feat(hook): honor before_processing return value to gate chain execution before_processing now follows the same continue/halt contract as chain links: a truthy return (vcon_id) means continue, a falsy return means skip the chain for this vCon without raising and without sending it to the DLQ. The default hook returns vcon_id so existing behaviour is preserved. The worker logs the skip and after_processing still runs. This gives hooks a graceful "defer" signal so they can manage their own retry/back-off state without forcing the worker to DLQ the vCon. Co-Authored-By: Claude Opus 4.7 (1M context) --- conserver/hook.py | 9 +++++++-- conserver/main.py | 7 ++++++- 2 files changed, 13 insertions(+), 3 deletions(-) diff --git a/conserver/hook.py b/conserver/hook.py index d275d7c4..519b7e8e 100644 --- a/conserver/hook.py +++ b/conserver/hook.py @@ -12,15 +12,20 @@ def before_processing( vcon_id: str, chain_details: Dict[str, Any], context: Optional[Dict[str, Any]], -) -> None: +) -> Optional[str]: """Called before processing of a vCon starts. Args: vcon_id: The vCon identifier. chain_details: The chain configuration for this processing run. context: Optional context data (e.g. trace_id, span_id). + + Returns: + The vCon identifier to continue chain processing, or a falsy value + (None / False / "") to skip processing for this vCon without raising + and without sending it to the DLQ. Mirrors the link contract. """ - pass + return vcon_id def after_processing( diff --git a/conserver/main.py b/conserver/main.py index d60acc0c..c014593f 100644 --- a/conserver/main.py +++ b/conserver/main.py @@ -732,7 +732,12 @@ def _handle_vcon( try: context = context or {} context["ingress_list"] = ingress_list - hook.before_processing(vcon_id, chain_details, context) + if not hook.before_processing(vcon_id, chain_details, context): + logger.info( + "[%s] before_processing skipped vCon %s on ingress %s", + worker_name, vcon_id, ingress_list, + ) + return vcon_chain_request.process() except Exception as e: processing_error = e