golden hour

/opt/imunify360/venv/versions/imunify-core-8.14.0-1/defence360agent/internals

File/Folder	Size	Actions
__init__.py	0 B
__pycache__	-
auth_protocol.py	1.18 KB
cln.py	13.59 KB
deadlock_detecting_lock.py	755 B
delivery_ack.py	1.67 KB
feature_flags.py	9.68 KB
geo.py	2.39 KB
global_scope.py	462 B
iaid.py	14.32 KB
lazy_load.py	195 B
logger.py	15.53 KB
logging_protocol.py	1.07 KB
message_status_publisher.py	6.74 KB
persistent_message.py	6.66 KB
the_sink.py	12.09 KB

Edit: delivery_ack.py

"""In-memory delivery acknowledgements for Reportable messages.

The send-to-server plugins queue messages rather than deliver them, so a
producer that keeps its own copy of the payload cannot tell a delivered
message from one a lost send round dropped. Every send path reports the ids
the transport accepted here, and producers register the ids they care about.

Acknowledgements are deliberately not persisted: one that never arrives
leaves the message unconfirmed and makes the producer send it again. That
costs a duplicate the server may well store twice, whereas a silently
dropped payload cannot be recovered at all.
"""

import logging
from typing import Callable, Optional

logger = logging.getLogger(__name__)

class DeliveryAckRegistry:
    def __init__(self) -> None:
        self._callbacks: dict[str, Callable[[], None]] = {}

def watch(self, message_id: str, on_delivered: Callable[[], None]) -> None:
        if not message_id:
            return
        self._callbacks[message_id] = on_delivered

def unwatch(self, message_id: str) -> None:
        self._callbacks.pop(message_id, None)

def confirm(self, message_id: Optional[str]) -> None:
        on_delivered = self._callbacks.pop(message_id, None)
        if on_delivered is None:
            return
        try:
            on_delivered()
        except Exception:
            # a producer's bookkeeping must never break a send round, but it
            # failing means the producer will re-send forever: log the
            # traceback, this is the only place that sees it
            logger.exception(
                "Delivery acknowledgement for %s failed", message_id
            )

registry = DeliveryAckRegistry()