Skip to content

symfonic.core.observability.metrics_quiescence

metrics_quiescence

Process-local callback barrier; durable epoch authority lives in the host.

MetricsQuiescenceMixin

close_tenant_admission 

close_tenant_admission(tenant_id)

Synchronously deny new callbacks and attempt bindings before waiting.

Source code in src/symfonic/core/observability/metrics_quiescence.py 
113
114
115
116
117
118
119
120
121
def close_tenant_admission(self, tenant_id):
    """Synchronously deny new callbacks and attempt bindings before waiting."""
    if not tenant_id:
        raise ValueError("tenant_id must be nonempty")
    barrier = self._callback_barrier
    with barrier.lock:
        barrier.closed.add(tenant_id)
        barrier.retired.update(key for key, tenant in self._current_tenant.items()
                               if tenant == tenant_id)

quiesce_tenant async 

quiesce_tenant(tenant_id)

Close admission, wait callbacks and sink writes, then purge this tenant.

Cancellation leaves admission closed. Retry must finish the wait and purge before the owning worker can acknowledge a durable epoch.

Source code in src/symfonic/core/observability/metrics_quiescence.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
async def quiesce_tenant(self, tenant_id):
    """Close admission, wait callbacks and sink writes, then purge this tenant.

    Cancellation leaves admission closed. Retry must finish the wait and
    purge before the owning worker can acknowledge a durable epoch.
    """
    self.close_tenant_admission(tenant_id)
    barrier = self._callback_barrier
    with barrier.lock:
        anonymous_keys = tuple(barrier.active_anonymous)
        for key in anonymous_keys:
            barrier.fenced_anonymous[key] = barrier.fenced_anonymous.get(key, 0) + 1
    try:
        await barrier.wait_anonymous(anonymous_keys)
        await self._drain_tenant(tenant_id)
    finally:
        with barrier.lock:
            for key in anonymous_keys:
                barrier.fenced_anonymous[key] -= 1
                if not barrier.fenced_anonymous[key]:
                    del barrier.fenced_anonymous[key]

guarded_binding 

guarded_binding(bind)

Keep validation and synchronous attribution atomic with admission close.

Source code in src/symfonic/core/observability/metrics_quiescence.py 
88
89
90
91
92
93
94
def guarded_binding(bind):
    """Keep validation and synchronous attribution atomic with admission close."""
    @wraps(bind)
    def guarded(self, *args, **kwargs):
        with self._callback_barrier.lock:
            return bind(self, *args, **kwargs)
    return guarded

guarded_callback 

guarded_callback(callback)

Cover the complete callback, including awaits in budget fan-out.

Source code in src/symfonic/core/observability/metrics_quiescence.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def guarded_callback(callback):
    """Cover the complete callback, including awaits in budget fan-out."""
    @wraps(callback)
    async def guarded(self, event):
        barrier = self._callback_barrier
        with barrier.lock:
            key = self._event_key(event)
            tenant_id = self._current_tenant.get(key, "")
            barrier.require_open(key, tenant_id)
            barrier.active[tenant_id] = barrier.active.get(tenant_id, 0) + 1
            if not tenant_id:
                barrier.active_anonymous[key] = barrier.active_anonymous.get(key, 0) + 1
        try:
            return await callback(self, event)
        finally:
            barrier.leave(tenant_id, key)
    return guarded