Skip to content

sb_lock

azure_bootstrap.sb_lock

Service Bus message-lock management.

AutoLockRenewer defaults are conservative; this wrapper bounds its lifetime and ensures .close() is called on every path. Swallows construction failures — the renewer is a defense against long-running handlers exceeding the lock duration, not a correctness gate.

Classes:

Name Description
ManagedLock

OO variant of :func:lock_for_process.

Functions:

Name Description
lock_for_process

Register the message with an AutoLockRenewer for the block's duration.

ManagedLock 

ManagedLock(receiver: Any, msg: Any, *, max_lock_renewal_seconds: int = DEFAULT_MAX_LOCK_RENEWAL_SECONDS)

OO variant of :func:lock_for_process.

Useful when callers need to extend the lock explicitly across multi-stage pipelines. Supports the context-manager protocol too.

Source code in azure_bootstrap/sb_lock/__init__.py 
77
78
79
80
81
82
83
84
85
86
87
def __init__(
    self,
    receiver: Any,
    msg: Any,
    *,
    max_lock_renewal_seconds: int = DEFAULT_MAX_LOCK_RENEWAL_SECONDS,
) -> None:
    self._receiver = receiver
    self._msg = msg
    self._max = max_lock_renewal_seconds
    self._renewer: Any = None

lock_for_process 

lock_for_process(receiver: Any, msg: Any, *, max_lock_renewal_seconds: int = DEFAULT_MAX_LOCK_RENEWAL_SECONDS) -> Generator[None, None, None]

Register the message with an AutoLockRenewer for the block's duration.

Source code in azure_bootstrap/sb_lock/__init__.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@contextmanager
def lock_for_process(
    receiver: Any,
    msg: Any,
    *,
    max_lock_renewal_seconds: int = DEFAULT_MAX_LOCK_RENEWAL_SECONDS,
) -> Generator[None, None, None]:
    """Register the message with an AutoLockRenewer for the block's duration."""
    renewer: Any = None
    try:
        renewer = _new_auto_lock_renewer()
        renewer.register(
            receiver,
            msg,
            max_lock_renewal_duration=max_lock_renewal_seconds,
        )
        bump_counter("sb_lock.renewer_started")
    except Exception:
        # Construction failure: the renewer is defensive — let processing
        # continue without it. The broker will redeliver if the lock expires.
        bump_counter("sb_lock.renewer_construction_failed")
        _logger.warning(
            "lock_for_process: AutoLockRenewer setup failed; proceeding without renewal",
            exc_info=True,
        )
        renewer = None
    try:
        yield
    finally:
        if renewer is not None:
            try:
                renewer.close()
            except Exception:
                bump_counter("sb_lock.close_failed")
                _logger.warning("lock_for_process: renewer.close() raised", exc_info=True)