refactor(core): implement deterministic async flush synchronization

Replace timing-based best-effort flushing with a robust synchronization
barrier using a dedicated _FlushSignal sentinel and threading events.
The AsyncLogger.flush method now enqueues a signal that blocks the
calling thread until the background worker processes all preceding log
entries, ensuring data integrity during shutdown

Co-authored-by: gitbot <gitbot@gossorg.in>

Author: IntegerAlex

CHANGELOG.md

@@ -7,7 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
-## [2.2.0] - 
+## [0.2.2] - 2026-03-
 
 ### Added
 - `LOG_LEVEL_DEBUG`, `LOG_LEVEL_INFO`, `LOG_LEVEL_WARNING`, `LOG_LEVEL_ERROR`, and `LOG_LEVEL_CRITICAL` named constants exported from the top-level package, replacing bare integer literals throughout the API.

kakashi/__init__.py

@@ -95,6 +95,10 @@
 __description__ = "Professional High-Performance Logging Library"
 __url__ = "https://github.com/kakashi/logging"
 
+# Backward-compatible aliases for callers using `kakashi.version`/`kakashi.author`.
+version = __version__
+author = __author__
+
 # ============================================================================
 # MAIN EXPORTS
 # ============================================================================
@@ -151,6 +155,8 @@
     # ---- VERSION AND METADATA ----
     "__version__",
     "__author__",
+    "version",
+    "author",
     "__description__",
     "__url__",
-]
+]

kakashi/core/logger.py

@@ -35,9 +35,6 @@
 LOG_LEVEL_ERROR = 40
 LOG_LEVEL_CRITICAL = 50
 
-# Wait time (seconds) for AsyncLogger.close() best-effort flush
-_ASYNC_CLOSE_WAIT_SECS = 0.05
-
 # Thread-local storage for lock-free operation
 _thread_local = threading.local()
 
@@ -47,44 +44,74 @@
 _async_shutdown = threading.Event()
 
 
+class _FlushSignal:
+    """Queue item used to signal a synchronous flush barrier."""
+
+    __slots__ = ("event",)
+
+    def __init__(self, event: threading.Event):
+        self.event = event
+
+
 
 
 
 def _async_worker_thread():
     """Background worker for async logging."""
     batch = []
     batch_size = 50  # Optimal batch size for throughput/latency balance
-    
-    while not _async_shutdown.is_set():
+
+    while True:
         try:
             # Collect batch
             batch.clear()
             timeout = 0.1  # 100ms batch timeout
-            
+            shutdown_requested = False
+            pending_count = 0
+
             try:
                 # Get first item (blocking)
                 item = _async_queue.get(timeout=timeout)
-                if item is None:  # Shutdown signal
+                if item is None:
+                    _async_queue.task_done()
                     break
+                if isinstance(item, _FlushSignal):
+                    item.event.set()
+                    _async_queue.task_done()
+                    continue
+
                 batch.append(item)
-                
+                pending_count = 1
+
                 # Collect additional items (non-blocking)
                 for _ in range(batch_size - 1):
                     try:
                         item = _async_queue.get_nowait()
-                        if item is None:  # Shutdown signal
+                        if item is None:
+                            _async_queue.task_done()
+                            shutdown_requested = True
                             break
+                        if isinstance(item, _FlushSignal):
+                            item.event.set()
+                            _async_queue.task_done()
+                            continue
                         batch.append(item)
+                        pending_count += 1
                     except queue.Empty:
                         break
-            
+
             except queue.Empty:
                 continue
-            
+
             # Process batch
             if batch:
                 _process_async_batch(batch)
-                
+                for _ in range(pending_count):
+                    _async_queue.task_done()
+
+            if shutdown_requested:
+                break
+
         except Exception:
             pass  # Ignore errors in background thread
 
@@ -255,12 +282,8 @@ def __init__(self, name: str, min_level: int = LOG_LEVEL_INFO):
         _ensure_async_worker()
 
     def close(self) -> None:
-        """Best-effort flush of pending messages for this logger by waiting briefly."""
-        # Draining items belonging to this logger from the queue would require
-        # restructuring the shared queue, so we do a best-effort flush by
-        # waiting briefly for the background worker to catch up. This does not
-        # guarantee that all enqueued messages have been processed.
-        time.sleep(_ASYNC_CLOSE_WAIT_SECS)
+        """Flush all pending async messages before returning."""
+        self.flush()
 
     def _log_async(self, level: int, message: str, fields: Optional[Dict[str, Any]] = None) -> None:
         """True asynchronous logging - non-blocking enqueue."""
@@ -310,11 +333,11 @@ def exception(self, message: str, **fields) -> None:
         self._log_async(40, message, fields)
 
     def flush(self) -> None:
-        """Best-effort, timing-based flush by yielding to the background worker."""
-        # For async logger, we can't force immediate flush, but we can yield
-        # briefly to allow background processing. This does not guarantee that
-        # all pending messages have been processed.
-        time.sleep(_ASYNC_CLOSE_WAIT_SECS)
+        """Block until all queued async messages are processed."""
+        _ensure_async_worker()
+        marker = _FlushSignal(threading.Event())
+        _async_queue.put(marker)
+        marker.event.wait()
 
 
 # Lock-free logger cache using thread-local storage
@@ -376,6 +399,9 @@ def shutdown_async_logging() -> None:
     """Shutdown async logging gracefully."""
     global _async_worker
     if _async_worker and _async_worker.is_alive():
+        # Ensure all enqueued items are processed before stopping the worker.
+        _async_queue.join()
+
         # Send shutdown sentinel so the worker can finish processing
         try:
             _async_queue.put_nowait(None)  # Shutdown signal

performance_tests/test_api_compatibility.py

@@ -11,6 +11,7 @@
 from pathlib import Path
 from typing import Any
 import tempfile
+import threading
 
 # Add parent directory to path
 sys.path.insert(0, str(Path(__file__).parent.parent))
@@ -240,3 +241,54 @@ def log_messages():
 
         # If we get here without errors, concurrent access works
         assert True
+
+
+class TestAsyncFlushRegression:
+    """Regression coverage for AsyncLogger.flush synchronization semantics."""
+
+    def test_async_flush_waits_for_worker_processing(self, monkeypatch):
+        """
+        Ensure flush is synchronization-based, not timing-based.
+
+        The worker is deliberately blocked while processing a batch. flush()
+        must block until processing is released.
+        """
+        from kakashi import get_async_logger, shutdown_async_logging
+        import kakashi.core.logger as core_logger
+
+        original_process = core_logger._process_async_batch
+        processing_started = threading.Event()
+        release_processing = threading.Event()
+        flush_finished = threading.Event()
+
+        def blocking_process(batch):
+            processing_started.set()
+            # If flush is truly synchronized with queue processing, it will
+            # remain blocked until this event is released.
+            release_processing.wait(timeout=2.0)
+            original_process(batch)
+
+        monkeypatch.setattr(core_logger, "_process_async_batch", blocking_process)
+
+        logger = get_async_logger("test_async_flush_regression")
+        logger.info("must be processed before flush returns")
+
+        # If flush regresses to a fixed sleep, this worker blocking won't matter.
+        flush_thread = threading.Thread(
+            target=lambda: (logger.flush(), flush_finished.set()),
+            daemon=True,
+        )
+        flush_thread.start()
+
+        assert processing_started.wait(timeout=1.0), "Worker never started processing"
+        assert not flush_finished.wait(timeout=0.05), (
+            "flush() returned before queued work finished; likely timing-based regression"
+        )
+
+        release_processing.set()
+        flush_thread.join(timeout=1.0)
+        assert not flush_thread.is_alive(), "flush() did not finish after worker was released"
+        assert flush_finished.is_set(), "flush() did not complete successfully"
+
+        # Keep test isolation strong for following tests.
+        shutdown_async_logging()