♻️ sync owner migration and facade thinning

src-async/BitLoggerAsync_test.mbt

@@ -25,7 +25,7 @@ async test "shutdown drains pending records" {
     target="async.test",
     flush=fn(_) {
       flushes.val += 1
-      1
+      ()
     },
   )
 
@@ -141,14 +141,84 @@ async test "closed logger runtime determines whether later log reaches patch pat
   inspect(logger.dropped_count(), content="1")
 
   logger.info("late")
-  inspect(
-    patched.val,
-    content=if async_runtime_supports_background_worker() { "2" } else { "1" },
-  )
+  inspect(patched.val, content="1")
   inspect(logger.pending_count(), content="0")
   inspect(logger.dropped_count(), content="1")
 }
 
+///|
+async test "closed logger does not enter filter path on later log attempts" {
+  let filtered : Ref[Int] = Ref(0)
+  let logger = async_logger(
+    @bitlogger.callback_sink(fn(_) {  }),
+    config=AsyncLoggerConfig::new(
+      max_pending=2,
+      overflow=AsyncOverflowPolicy::Blocking,
+    ),
+    min_level=@bitlogger.Level::Info,
+    target="async.closed.filter",
+  ).with_filter(fn(_) {
+    filtered.val += 1
+    true
+  })
+
+  logger.info("one")
+  inspect(filtered.val, content="1")
+  logger.close(clear=true)
+
+  logger.info("late")
+  inspect(filtered.val, content="1")
+  inspect(logger.pending_count(), content="0")
+  inspect(logger.dropped_count(), content="1")
+}
+
+///|
+async test "run rejects duplicate worker startup for same logger" {
+  let logger = async_logger(
+    @bitlogger.callback_sink(fn(_) {  }),
+    config=AsyncLoggerConfig::new(
+      max_pending=2,
+      overflow=AsyncOverflowPolicy::Blocking,
+    ),
+    min_level=@bitlogger.Level::Info,
+    target="async.run.duplicate",
+  )
+
+  @async.with_task_group(group => {
+    group.spawn_bg(() => logger.run())
+    while !logger.is_running() {
+      @async.pause()
+    }
+
+    let second_error = (async fn() -> String raise {
+      logger.run()
+      "no error"
+    })() catch {
+      err => err.to_string()
+    }
+
+    inspect(second_error.contains("AsyncLoggerAlreadyRunning"), content="true")
+    let state = logger.state()
+    inspect(
+      match state.phase {
+        AsyncLifecyclePhase::Running => "running"
+        _ => "other"
+      },
+      content="running",
+    )
+    inspect(logger.has_failed(), content="false")
+    inspect(logger.last_error(), content="")
+    inspect(state.can_rerun, content="false")
+    inspect(state.terminal, content="false")
+    logger.shutdown(clear=true)
+  })
+
+  inspect(logger.is_closed(), content="true")
+  inspect(logger.is_running(), content="false")
+  inspect(logger.has_failed(), content="false")
+  inspect(logger.last_error(), content="")
+}
+
 ///|
 async test "async logger with_filter composes and leaves base logger unchanged" {
   let written_targets : Ref[Array[String]] = Ref([])
@@ -1128,9 +1198,23 @@ test "async logger state snapshot reflects current counters and runtime" {
   )
   inspect(state.pending_count, content="0")
   inspect(state.dropped_count, content="0")
+  inspect(
+    match state.phase {
+      AsyncLifecyclePhase::Ready => "ready"
+      AsyncLifecyclePhase::Running => "running"
+      AsyncLifecyclePhase::Failed => "failed"
+      AsyncLifecyclePhase::Closing => "closing"
+      AsyncLifecyclePhase::Closed => "closed"
+      AsyncLifecyclePhase::ClosedFailed => "closed_failed"
+    },
+    content="ready",
+  )
   inspect(state.is_closed, content="false")
   inspect(state.is_running, content="false")
   inspect(state.has_failed, content="false")
+  inspect(state.backlog_retained, content="false")
+  inspect(state.can_rerun, content="true")
+  inspect(state.terminal, content="false")
   inspect(state.last_error, content="")
   inspect(
     match state.flush_policy {
@@ -1143,21 +1227,54 @@ test "async logger state snapshot reflects current counters and runtime" {
   inspect(
     @json_parser.stringify(async_logger_state_to_json(state)),
     content=if async_runtime_supports_background_worker() {
-      "{\"runtime\":{\"mode\":\"native_worker\",\"background_worker\":true},\"pending_count\":0,\"dropped_count\":0,\"is_closed\":false,\"is_running\":false,\"has_failed\":false,\"last_error\":\"\",\"flush_policy\":\"Shutdown\"}"
+      "{\"runtime\":{\"mode\":\"native_worker\",\"background_worker\":true},\"phase\":\"ready\",\"pending_count\":0,\"dropped_count\":0,\"is_closed\":false,\"is_running\":false,\"has_failed\":false,\"backlog_retained\":false,\"can_rerun\":true,\"terminal\":false,\"last_error\":\"\",\"flush_policy\":\"Shutdown\"}"
     } else {
-      "{\"runtime\":{\"mode\":\"compatibility\",\"background_worker\":false},\"pending_count\":0,\"dropped_count\":0,\"is_closed\":false,\"is_running\":false,\"has_failed\":false,\"last_error\":\"\",\"flush_policy\":\"Shutdown\"}"
+      "{\"runtime\":{\"mode\":\"compatibility\",\"background_worker\":false},\"phase\":\"ready\",\"pending_count\":0,\"dropped_count\":0,\"is_closed\":false,\"is_running\":false,\"has_failed\":false,\"backlog_retained\":false,\"can_rerun\":true,\"terminal\":false,\"last_error\":\"\",\"flush_policy\":\"Shutdown\"}"
     },
   )
   inspect(
     stringify_async_logger_state(state),
     content=if async_runtime_supports_background_worker() {
-      "{\"runtime\":{\"mode\":\"native_worker\",\"background_worker\":true},\"pending_count\":0,\"dropped_count\":0,\"is_closed\":false,\"is_running\":false,\"has_failed\":false,\"last_error\":\"\",\"flush_policy\":\"Shutdown\"}"
+      "{\"runtime\":{\"mode\":\"native_worker\",\"background_worker\":true},\"phase\":\"ready\",\"pending_count\":0,\"dropped_count\":0,\"is_closed\":false,\"is_running\":false,\"has_failed\":false,\"backlog_retained\":false,\"can_rerun\":true,\"terminal\":false,\"last_error\":\"\",\"flush_policy\":\"Shutdown\"}"
     } else {
-      "{\"runtime\":{\"mode\":\"compatibility\",\"background_worker\":false},\"pending_count\":0,\"dropped_count\":0,\"is_closed\":false,\"is_running\":false,\"has_failed\":false,\"last_error\":\"\",\"flush_policy\":\"Shutdown\"}"
+      "{\"runtime\":{\"mode\":\"compatibility\",\"background_worker\":false},\"phase\":\"ready\",\"pending_count\":0,\"dropped_count\":0,\"is_closed\":false,\"is_running\":false,\"has_failed\":false,\"backlog_retained\":false,\"can_rerun\":true,\"terminal\":false,\"last_error\":\"\",\"flush_policy\":\"Shutdown\"}"
     },
   )
 }
 
+///|
+async test "async logger state enters running phase while worker is active" {
+  let logger = async_logger(
+    @bitlogger.callback_sink(fn(_) {  }),
+    config=AsyncLoggerConfig::new(max_pending=2),
+    min_level=@bitlogger.Level::Info,
+    target="async.state.running",
+  )
+
+  @async.with_task_group(group => {
+    group.spawn_bg(() => logger.run())
+    while !logger.is_running() {
+      @async.pause()
+    }
+
+    let state = logger.state()
+    inspect(
+      match state.phase {
+        AsyncLifecyclePhase::Running => "running"
+        _ => "other"
+      },
+      content="running",
+    )
+    inspect(state.is_closed, content="false")
+    inspect(state.is_running, content="true")
+    inspect(state.has_failed, content="false")
+    inspect(state.backlog_retained, content="false")
+    inspect(state.can_rerun, content="false")
+    inspect(state.terminal, content="false")
+    logger.shutdown(clear=true)
+  })
+}
+
 ///|
 async test "run drains queued records in compatibility backends too" {
   let written : Ref[Array[String]] = Ref([])
@@ -1201,7 +1318,7 @@ async test "async logger records worker failures and wait_idle stops early" {
     ),
     min_level=@bitlogger.Level::Info,
     target="async.failure",
-    flush=fn(_) -> Int raise { raise TestFlushError("flush exploded") },
+    flush=fn(_) -> Unit raise { raise TestFlushError("flush exploded") },
   )
 
   @async.with_task_group(group => {
@@ -1219,6 +1336,18 @@ async test "async logger records worker failures and wait_idle stops early" {
   inspect(logger.last_error().contains("TestFlushError"), content="true")
   inspect(logger.is_running(), content="false")
   inspect(logger.pending_count(), content="0")
+
+  let state = logger.state()
+  inspect(
+    match state.phase {
+      AsyncLifecyclePhase::ClosedFailed => "closed_failed"
+      _ => "other"
+    },
+    content="closed_failed",
+  )
+  inspect(state.backlog_retained, content="false")
+  inspect(state.can_rerun, content="false")
+  inspect(state.terminal, content="true")
 }
 
 ///|
@@ -1234,12 +1363,12 @@ async test "later started run resets async failure state before draining remaini
     ),
     min_level=@bitlogger.Level::Info,
     target="async.retry",
-    flush=fn(_) -> Int raise {
+    flush=fn(_) -> Unit raise {
       flushes.val += 1
       if flushes.val == 1 {
         raise TestFlushError("flush exploded once")
       }
-      1
+      ()
     },
   )
 
@@ -1253,12 +1382,33 @@ async test "later started run resets async failure state before draining remaini
     inspect(logger.is_running(), content="false")
     inspect(logger.pending_count(), content="1")
 
+    let failed_state = logger.state()
+    inspect(
+      match failed_state.phase {
+        AsyncLifecyclePhase::Failed => "failed"
+        _ => "other"
+      },
+      content="failed",
+    )
+    inspect(failed_state.backlog_retained, content="true")
+    inspect(failed_state.can_rerun, content="true")
+    inspect(failed_state.terminal, content="false")
+
     group.spawn_bg(() => logger.run())
     while !logger.is_running() {
       @async.pause()
     }
     inspect(logger.has_failed(), content="false")
     inspect(logger.last_error(), content="")
+
+    let running_state = logger.state()
+    inspect(
+      match running_state.phase {
+        AsyncLifecyclePhase::Running => "running"
+        _ => "other"
+      },
+      content="running",
+    )
     logger.shutdown()
   })
 
@@ -1272,7 +1422,7 @@ async test "later started run resets async failure state before draining remaini
 }
 
 ///|
-async test "shutdown after worker failure uses runtime-specific pending cleanup" {
+async test "shutdown after worker failure clears remaining pending backlog" {
   let writes : Ref[Int] = Ref(0)
   let logger = async_logger(
     @bitlogger.callback_sink(fn(_) { writes.val += 1 }),
@@ -1283,7 +1433,7 @@ async test "shutdown after worker failure uses runtime-specific pending cleanup"
     ),
     min_level=@bitlogger.Level::Info,
     target="async.failure.shutdown",
-    flush=fn(_) -> Int raise {
+    flush=fn(_) -> Unit raise {
       raise TestFlushError("flush exploded on shutdown path")
     },
   )
@@ -1303,14 +1453,58 @@ async test "shutdown after worker failure uses runtime-specific pending cleanup"
   inspect(logger.last_error().contains("TestFlushError"), content="true")
   inspect(logger.is_running(), content="false")
   inspect(writes.val, content="1")
+  inspect(logger.pending_count(), content="0")
+  inspect(logger.dropped_count(), content="1")
+
+  let state = logger.state()
   inspect(
-    logger.pending_count(),
-    content=if async_runtime_supports_background_worker() { "0" } else { "1" },
+    match state.phase {
+      AsyncLifecyclePhase::ClosedFailed => "closed_failed"
+      _ => "other"
+    },
+    content="closed_failed",
   )
+  inspect(state.is_closed, content="true")
+  inspect(state.is_running, content="false")
+  inspect(state.has_failed, content="true")
+  inspect(state.pending_count, content="0")
+  inspect(state.dropped_count, content="1")
+  inspect(state.backlog_retained, content="false")
+  inspect(state.can_rerun, content="false")
+  inspect(state.terminal, content="true")
+}
+
+///|
+async test "shutdown clear reaches terminal closed phase without worker startup" {
+  let logger = async_logger(
+    @bitlogger.callback_sink(fn(_) {  }),
+    config=AsyncLoggerConfig::new(
+      max_pending=2,
+      overflow=AsyncOverflowPolicy::Blocking,
+    ),
+    min_level=@bitlogger.Level::Info,
+    target="async.shutdown.clear.state",
+  )
+
+  logger.info("one")
+  logger.shutdown(clear=true)
+
+  let state = logger.state()
   inspect(
-    logger.dropped_count(),
-    content=if async_runtime_supports_background_worker() { "1" } else { "0" },
+    match state.phase {
+      AsyncLifecyclePhase::Closed => "closed"
+      _ => "other"
+    },
+    content="closed",
   )
+  inspect(state.is_closed, content="true")
+  inspect(state.is_running, content="false")
+  inspect(state.has_failed, content="false")
+  inspect(state.pending_count, content="0")
+  inspect(state.dropped_count, content="1")
+  inspect(state.backlog_retained, content="false")
+  inspect(state.can_rerun, content="false")
+  inspect(state.terminal, content="true")
 }
 
 ///|
@@ -1358,7 +1552,7 @@ async test "library async new preserves config and flush failure contract" {
     ),
     min_level=@bitlogger.Level::Warn,
     target="async.lib.new",
-    flush=fn(_) -> Int raise {
+    flush=fn(_) -> Unit raise {
       flushes.val += 1
       raise TestFlushError("library new flush exploded")
     },
@@ -1687,7 +1881,7 @@ async test "library async shutdown preserves wrapped failure cleanup semantics"
     ),
     min_level=@bitlogger.Level::Info,
     target="async.lib.failure",
-    flush=fn(_) -> Int raise {
+    flush=fn(_) -> Unit raise {
       raise TestFlushError("library facade flush exploded")
     },
   )
@@ -2847,6 +3041,47 @@ async test "build async logger keeps direct helper surface and runtime sink path
   inspect(logger.dropped_count(), content="1")
   inspect(logger.has_failed(), content="false")
   inspect(logger.last_error(), content="")
+  inspect(logger.sink.pending_count(), content="0")
+}
+
+///|
+async test "build async logger batch flush policy drains queued runtime sink work at batch boundary" {
+  let logger = build_async_logger(
+    AsyncLoggerBuildConfig::new(
+      logger=@bitlogger.LoggerConfig::new(
+        min_level=@bitlogger.Level::Info,
+        target="async.direct.batch.flush.runtime",
+        queue=Some(
+          @bitlogger.QueueConfig::new(
+            4,
+            overflow=@bitlogger.QueueOverflowPolicy::DropNewest,
+          ),
+        ),
+        sink=@bitlogger.SinkConfig::new(kind=@bitlogger.SinkKind::Console),
+      ),
+      async_config=AsyncLoggerConfig::new(
+        max_pending=4,
+        overflow=AsyncOverflowPolicy::Blocking,
+        max_batch=2,
+        flush=AsyncFlushPolicy::Batch,
+      ),
+    ),
+  )
+
+  @async.with_task_group(group => {
+    group.spawn_bg(() => logger.run())
+    logger.info("one")
+    logger.info("two")
+    logger.wait_idle()
+    inspect(logger.pending_count(), content="0")
+    inspect(logger.sink.pending_count(), content="0")
+    inspect(logger.has_failed(), content="false")
+    logger.shutdown()
+  })
+
+  inspect(logger.is_closed(), content="true")
+  inspect(logger.is_running(), content="false")
+  inspect(logger.sink.pending_count(), content="0")
 }
 
 ///|
@@ -2959,7 +3194,7 @@ async test "async logger projection preserves async queue and failure state" {
     ),
     min_level=@bitlogger.Level::Info,
     target="async.projected.state",
-    flush=fn(_) -> Int raise {
+    flush=fn(_) -> Unit raise {
       raise TestFlushError("projected facade flush exploded")
     },
   ).to_library_async_logger()
@@ -4471,7 +4706,7 @@ async test "build async text logger batch flush policy keeps default no-op flush
     config=config.async_config,
     min_level=config.logger.min_level,
     target=config.logger.target,
-    flush=fn(_) -> Int raise {
+    flush=fn(_) -> Unit raise {
       raise TestFlushError("text batch flush exploded")
     },
   ).with_timestamp(enabled=config.logger.timestamp)
@@ -4556,7 +4791,7 @@ async test "build async text logger shutdown flush policy keeps default no-op fl
     config=config.async_config,
     min_level=config.logger.min_level,
     target=config.logger.target,
-    flush=fn(_) -> Int raise {
+    flush=fn(_) -> Unit raise {
       raise TestFlushError("text shutdown flush exploded")
     },
   ).with_timestamp(enabled=config.logger.timestamp)

src-async/async_logger_native.mbt

@@ -8,18 +8,3 @@ pub fn async_runtime_supports_background_worker() -> Bool {
   ignore(all_async_runtime_modes())
   true
 }
-
-///|
-fn async_runtime_guard_closed_on_log() -> Bool {
-  false
-}
-
-///|
-fn async_runtime_shutdown_clears_pending_after_wait_idle() -> Bool {
-  true
-}
-
-///|
-fn async_runtime_shutdown_waits_for_worker() -> Bool {
-  true
-}

src-async/async_logger_shared.mbt

@@ -3,6 +3,11 @@ pub(all) suberror AsyncLoggerClosed {
   AsyncLoggerClosed
 }
 
+///|
+pub(all) suberror AsyncLoggerAlreadyRunning {
+  AsyncLoggerAlreadyRunning
+}
+
 ///|
 pub type AsyncOverflowPolicy = @utils.AsyncOverflowPolicy
 
@@ -23,6 +28,9 @@ fn all_async_runtime_modes() -> Array[AsyncRuntimeMode] {
 ///|
 pub type AsyncRuntimeState = @utils.AsyncRuntimeState
 
+///|
+pub type AsyncLifecyclePhase = @utils.AsyncLifecyclePhase
+
 ///|
 pub type AsyncLoggerState = @utils.AsyncLoggerState
 
@@ -129,16 +137,14 @@ pub struct AsyncLogger[S] {
   linger_ms : Int
   flush_policy : AsyncFlushPolicy
   sink : S
-  flush_sink : (S) -> Int raise
+  flush_callback : (S) -> Unit raise
   context_fields : Array[@bitlogger.Field]
   filter : (@bitlogger.Record) -> Bool
   patch : @bitlogger.RecordPatch
   queue : @async.Queue[@bitlogger.Record]
   pending_count : Ref[Int]
   dropped_count : Ref[Int]
-  is_closed : Ref[Bool]
-  is_running : Ref[Bool]
-  has_failed : Ref[Bool]
+  phase : Ref[AsyncLifecyclePhase]
   last_error : Ref[String]
 }
 
@@ -148,7 +154,7 @@ pub fn[S] async_logger(
   config? : AsyncLoggerConfig = AsyncLoggerConfig::new(),
   min_level? : @bitlogger.Level = @bitlogger.Level::Info,
   target? : String = "",
-  flush? : (S) -> Int raise = fn(_) { 0 },
+  flush? : (S) -> Unit raise = fn(_) { () },
 ) -> AsyncLogger[S] {
   {
     min_level,
@@ -159,20 +165,87 @@ pub fn[S] async_logger(
     linger_ms: config.linger_ms,
     flush_policy: config.flush,
     sink,
-    flush_sink: flush,
+    flush_callback: flush,
     context_fields: [],
     filter: fn(_) { true },
     patch: @bitlogger.identity_patch(),
     queue: @async.Queue(kind=queue_kind_of(config)),
     pending_count: Ref(0),
     dropped_count: Ref(0),
-    is_closed: Ref(false),
-    is_running: Ref(false),
-    has_failed: Ref(false),
+    phase: Ref(AsyncLifecyclePhase::Ready),
     last_error: Ref(""),
   }
 }
 
+///|
+fn async_logger_phase_is_closed(phase : AsyncLifecyclePhase) -> Bool {
+  match phase {
+    AsyncLifecyclePhase::Closing |
+    AsyncLifecyclePhase::Closed |
+    AsyncLifecyclePhase::ClosedFailed => true
+    _ => false
+  }
+}
+
+///|
+fn async_logger_phase_is_running(phase : AsyncLifecyclePhase) -> Bool {
+  match phase {
+    AsyncLifecyclePhase::Running | AsyncLifecyclePhase::Closing => true
+    _ => false
+  }
+}
+
+///|
+fn async_logger_phase_has_failed(phase : AsyncLifecyclePhase) -> Bool {
+  match phase {
+    AsyncLifecyclePhase::Failed | AsyncLifecyclePhase::ClosedFailed => true
+    _ => false
+  }
+}
+
+///|
+fn async_logger_phase_after_close(
+  phase : AsyncLifecyclePhase,
+) -> AsyncLifecyclePhase {
+  match phase {
+    AsyncLifecyclePhase::Running | AsyncLifecyclePhase::Closing =>
+      AsyncLifecyclePhase::Closing
+    AsyncLifecyclePhase::Failed | AsyncLifecyclePhase::ClosedFailed =>
+      AsyncLifecyclePhase::ClosedFailed
+    AsyncLifecyclePhase::Closed => AsyncLifecyclePhase::Closed
+    AsyncLifecyclePhase::Ready => AsyncLifecyclePhase::Closed
+  }
+}
+
+///|
+fn async_logger_phase_after_run_exit(
+  phase : AsyncLifecyclePhase,
+) -> AsyncLifecyclePhase {
+  match phase {
+    AsyncLifecyclePhase::Closing => AsyncLifecyclePhase::Closed
+    AsyncLifecyclePhase::Running => AsyncLifecyclePhase::Ready
+    AsyncLifecyclePhase::Closed | AsyncLifecyclePhase::ClosedFailed => phase
+    _ => phase
+  }
+}
+
+///|
+fn async_logger_phase_after_run_failure(
+  phase : AsyncLifecyclePhase,
+) -> AsyncLifecyclePhase {
+  match phase {
+    AsyncLifecyclePhase::Closing => AsyncLifecyclePhase::ClosedFailed
+    AsyncLifecyclePhase::Closed | AsyncLifecyclePhase::ClosedFailed =>
+      AsyncLifecyclePhase::ClosedFailed
+    _ => AsyncLifecyclePhase::Failed
+  }
+}
+
+///|
+pub fn[S] AsyncLogger::phase(self : AsyncLogger[S]) -> AsyncLifecyclePhase {
+  self.phase.val
+}
+
 ///|
 fn queue_kind_of(config : AsyncLoggerConfig) -> @aqueue.Kind {
   let limit = if config.max_pending < 0 { 0 } else { config.max_pending }
@@ -282,7 +355,7 @@ pub async fn[S] AsyncLogger::log(
   fields? : Array[@bitlogger.Field] = [],
   target? : String = "",
 ) -> Unit {
-  guard !(async_runtime_guard_closed_on_log() && self.is_closed()) else { () }
+  guard !self.is_closed() else { () }
   guard self.is_enabled(level) else { () }
   let actual_target = if target == "" { self.target } else { target }
   let timestamp_ms = if self.timestamp { @env.now() } else { 0UL }
@@ -378,17 +451,17 @@ pub fn[S] AsyncLogger::dropped_count(self : AsyncLogger[S]) -> Int {
 
 ///|
 pub fn[S] AsyncLogger::is_closed(self : AsyncLogger[S]) -> Bool {
-  self.is_closed.val
+  async_logger_phase_is_closed(self.phase())
 }
 
 ///|
 pub fn[S] AsyncLogger::is_running(self : AsyncLogger[S]) -> Bool {
-  self.is_running.val
+  async_logger_phase_is_running(self.phase())
 }
 
 ///|
 pub fn[S] AsyncLogger::has_failed(self : AsyncLogger[S]) -> Bool {
-  self.has_failed.val
+  async_logger_phase_has_failed(self.phase())
 }
 
 ///|
@@ -405,11 +478,9 @@ pub fn[S] AsyncLogger::flush_policy(self : AsyncLogger[S]) -> AsyncFlushPolicy {
 pub fn[S] AsyncLogger::state(self : AsyncLogger[S]) -> AsyncLoggerState {
   AsyncLoggerState::new(
     async_runtime_state(),
+    self.phase(),
     self.pending_count(),
     self.dropped_count(),
-    self.is_closed(),
-    self.is_running(),
-    self.has_failed(),
     self.last_error(),
     self.flush_policy(),
   )
@@ -420,7 +491,7 @@ pub fn[S] AsyncLogger::close(
   self : AsyncLogger[S],
   clear? : Bool = false,
 ) -> Unit {
-  self.is_closed.val = true
+  self.phase.val = async_logger_phase_after_close(self.phase())
   if clear {
     let abandoned = self.pending_count()
     if abandoned > 0 {
@@ -450,20 +521,22 @@ pub async fn[S] AsyncLogger::shutdown(
     self.close(clear=true)
   } else {
     self.wait_idle()
-    if async_runtime_shutdown_clears_pending_after_wait_idle() &&
-      self.pending_count() > 0 {
+    if self.pending_count() > 0 {
       self.close(clear=true)
     } else {
       self.close()
     }
   }
-  if async_runtime_shutdown_waits_for_worker() {
-    while self.is_running() {
-      @async.pause()
-    }
+  while self.is_running() {
+    @async.pause()
   }
 }
 
+///|
+fn[S] run_flush_callback(logger : AsyncLogger[S]) -> Unit raise {
+  (logger.flush_callback)(logger.sink)
+}
+
 ///|
 async fn[S : @bitlogger.Sink] run_worker(logger : AsyncLogger[S]) -> Unit {
   while true {
@@ -512,12 +585,12 @@ async fn[S : @bitlogger.Sink] run_worker(logger : AsyncLogger[S]) -> Unit {
       }
     }
     match logger.flush_policy {
-      AsyncFlushPolicy::Batch => ignore((logger.flush_sink)(logger.sink))
+      AsyncFlushPolicy::Batch => run_flush_callback(logger)
       _ => ()
     }
   }
   match logger.flush_policy {
-    AsyncFlushPolicy::Shutdown => ignore((logger.flush_sink)(logger.sink))
+    AsyncFlushPolicy::Shutdown => run_flush_callback(logger)
     _ => ()
   }
 }
@@ -526,18 +599,18 @@ async fn[S : @bitlogger.Sink] run_worker(logger : AsyncLogger[S]) -> Unit {
 pub async fn[S : @bitlogger.Sink] AsyncLogger::run(
   self : AsyncLogger[S],
 ) -> Unit {
-  self.is_running.val = true
-  self.has_failed.val = false
+  guard !self.is_closed() else { raise AsyncLoggerClosed }
+  guard !self.is_running() else { raise AsyncLoggerAlreadyRunning }
+  self.phase.val = AsyncLifecyclePhase::Running
   self.last_error.val = ""
   run_worker(self) catch {
     err => {
-      self.has_failed.val = true
+      self.phase.val = async_logger_phase_after_run_failure(self.phase())
       self.last_error.val = err.to_string()
-      self.is_running.val = false
       raise err
     }
   }
-  self.is_running.val = false
+  self.phase.val = async_logger_phase_after_run_exit(self.phase())
 }
 
 ///|
@@ -550,7 +623,7 @@ pub fn build_async_logger(
     config=config.async_config,
     min_level=logger.min_level,
     target=logger.target,
-    flush=fn(sink) { sink.flush() },
+    flush=fn(sink) { ignore(sink.flush_progress()) },
   ).with_timestamp(enabled=logger.timestamp)
 }
 

src-async/async_logger_stub.mbt

@@ -8,18 +8,3 @@ pub fn async_runtime_supports_background_worker() -> Bool {
   ignore(all_async_runtime_modes())
   false
 }
-
-///|
-fn async_runtime_guard_closed_on_log() -> Bool {
-  true
-}
-
-///|
-fn async_runtime_shutdown_clears_pending_after_wait_idle() -> Bool {
-  false
-}
-
-///|
-fn async_runtime_shutdown_waits_for_worker() -> Bool {
-  false
-}

src-async/library_async_logger.mbt

@@ -21,7 +21,7 @@ pub fn[S] LibraryAsyncLogger::new(
   config? : AsyncLoggerConfig = AsyncLoggerConfig::new(),
   min_level? : @bitlogger.Level = @bitlogger.Level::Info,
   target? : String = "",
-  flush? : (S) -> Int raise = fn(_) { 0 },
+  flush? : (S) -> Unit raise = fn(_) { () },
 ) -> LibraryAsyncLogger[S] {
   library_async_logger(async_logger(sink, config~, min_level~, target~, flush~))
 }

src-async/pkg.generated.mbti

@@ -0,0 +1,157 @@
+// Generated using `moon info`, DON'T EDIT IT
+package "Nanaloveyuki/BitLogger/src-async"
+
+import {
+  "Nanaloveyuki/BitLogger/src",
+  "Nanaloveyuki/BitLogger/src-async/utils",
+  "Nanaloveyuki/BitLogger/src/core",
+  "maria/json_parser",
+  "moonbitlang/async/aqueue",
+  "moonbitlang/core/ref",
+}
+
+// Values
+pub fn[S] async_logger(S, config? : @utils.AsyncLoggerConfig, min_level? : @core.Level, target? : String, flush? : (S) -> Unit raise) -> AsyncLogger[S]
+
+pub fn async_logger_build_config_to_json(@utils.AsyncLoggerBuildConfig) -> @json_parser.JsonValue
+
+pub fn async_logger_config_to_json(@utils.AsyncLoggerConfig) -> @json_parser.JsonValue
+
+pub fn async_logger_state_to_json(@utils.AsyncLoggerState) -> @json_parser.JsonValue
+
+pub fn async_runtime_mode() -> @utils.AsyncRuntimeMode
+
+pub fn async_runtime_mode_label(@utils.AsyncRuntimeMode) -> String
+
+pub fn async_runtime_state() -> @utils.AsyncRuntimeState
+
+pub fn async_runtime_state_to_json(@utils.AsyncRuntimeState) -> @json_parser.JsonValue
+
+pub fn async_runtime_supports_background_worker() -> Bool
+
+pub fn build_application_async_logger(@utils.AsyncLoggerBuildConfig) -> AsyncLogger[@src.RuntimeSink]
+
+pub fn build_application_text_async_logger(@utils.AsyncLoggerBuildConfig) -> AsyncLogger[@src.FormattedConsoleSink]
+
+pub fn build_async_logger(@utils.AsyncLoggerBuildConfig) -> AsyncLogger[@src.RuntimeSink]
+
+pub fn build_async_text_logger(@utils.AsyncLoggerBuildConfig) -> AsyncLogger[@src.FormattedConsoleSink]
+
+pub fn build_library_async_logger(@utils.AsyncLoggerBuildConfig) -> LibraryAsyncLogger[@src.RuntimeSink]
+
+pub fn build_library_async_text_logger(@utils.AsyncLoggerBuildConfig) -> LibraryAsyncLogger[@src.FormattedConsoleSink]
+
+pub fn parse_and_build_application_async_logger(String) -> AsyncLogger[@src.RuntimeSink] raise
+
+pub fn parse_and_build_library_async_logger(String) -> LibraryAsyncLogger[@src.RuntimeSink] raise
+
+pub fn parse_async_logger_build_config_text(String) -> @utils.AsyncLoggerBuildConfig raise
+
+pub fn parse_async_logger_config_text(String) -> @utils.AsyncLoggerConfig raise
+
+pub fn stringify_async_logger_build_config(@utils.AsyncLoggerBuildConfig, pretty? : Bool) -> String
+
+pub fn stringify_async_logger_config(@utils.AsyncLoggerConfig, pretty? : Bool) -> String
+
+pub fn stringify_async_logger_state(@utils.AsyncLoggerState, pretty? : Bool) -> String
+
+pub fn stringify_async_runtime_state(@utils.AsyncRuntimeState, pretty? : Bool) -> String
+
+// Errors
+pub(all) suberror AsyncLoggerAlreadyRunning {
+  AsyncLoggerAlreadyRunning
+}
+
+pub(all) suberror AsyncLoggerClosed {
+  AsyncLoggerClosed
+}
+
+// Types and methods
+pub struct AsyncLogger[S] {
+  min_level : @core.Level
+  target : String
+  timestamp : Bool
+  overflow : @utils.AsyncOverflowPolicy
+  max_batch : Int
+  linger_ms : Int
+  flush_policy : @utils.AsyncFlushPolicy
+  sink : S
+  flush_callback : (S) -> Unit raise
+  context_fields : Array[@core.Field]
+  filter : (@core.Record) -> Bool
+  patch : (@core.Record) -> @core.Record
+  queue : @aqueue.Queue[@core.Record]
+  pending_count : @ref.Ref[Int]
+  dropped_count : @ref.Ref[Int]
+  phase : @ref.Ref[@utils.AsyncLifecyclePhase]
+  last_error : @ref.Ref[String]
+}
+pub fn[S] AsyncLogger::child(Self[S], String) -> Self[S]
+pub fn[S] AsyncLogger::close(Self[S], clear? : Bool) -> Unit
+pub async fn[S] AsyncLogger::debug(Self[S], String, fields? : Array[@core.Field]) -> Unit
+pub fn[S] AsyncLogger::dropped_count(Self[S]) -> Int
+pub async fn[S] AsyncLogger::error(Self[S], String, fields? : Array[@core.Field]) -> Unit
+pub fn[S] AsyncLogger::flush_policy(Self[S]) -> @utils.AsyncFlushPolicy
+pub fn[S] AsyncLogger::has_failed(Self[S]) -> Bool
+pub async fn[S] AsyncLogger::info(Self[S], String, fields? : Array[@core.Field]) -> Unit
+pub fn[S] AsyncLogger::is_closed(Self[S]) -> Bool
+pub fn[S] AsyncLogger::is_enabled(Self[S], @core.Level) -> Bool
+pub fn[S] AsyncLogger::is_running(Self[S]) -> Bool
+pub fn[S] AsyncLogger::last_error(Self[S]) -> String
+pub async fn[S] AsyncLogger::log(Self[S], @core.Level, String, fields? : Array[@core.Field], target? : String) -> Unit
+pub fn[S] AsyncLogger::pending_count(Self[S]) -> Int
+pub fn[S] AsyncLogger::phase(Self[S]) -> @utils.AsyncLifecyclePhase
+pub async fn[S : @src.Sink] AsyncLogger::run(Self[S]) -> Unit
+pub async fn[S] AsyncLogger::shutdown(Self[S], clear? : Bool) -> Unit
+pub fn[S] AsyncLogger::state(Self[S]) -> @utils.AsyncLoggerState
+pub fn[S] AsyncLogger::to_library_async_logger(Self[S]) -> LibraryAsyncLogger[S]
+pub async fn[S] AsyncLogger::trace(Self[S], String, fields? : Array[@core.Field]) -> Unit
+pub async fn[S] AsyncLogger::wait_idle(Self[S]) -> Unit
+pub async fn[S] AsyncLogger::warn(Self[S], String, fields? : Array[@core.Field]) -> Unit
+pub fn[S] AsyncLogger::with_context_fields(Self[S], Array[@core.Field]) -> Self[S]
+pub fn[S] AsyncLogger::with_filter(Self[S], (@core.Record) -> Bool) -> Self[S]
+pub fn[S] AsyncLogger::with_min_level(Self[S], @core.Level) -> Self[S]
+pub fn[S] AsyncLogger::with_patch(Self[S], (@core.Record) -> @core.Record) -> Self[S]
+pub fn[S] AsyncLogger::with_target(Self[S], String) -> Self[S]
+pub fn[S] AsyncLogger::with_timestamp(Self[S], enabled? : Bool) -> Self[S]
+
+pub struct LibraryAsyncLogger[S] {
+  inner : AsyncLogger[S]
+}
+pub fn[S] LibraryAsyncLogger::bind(Self[S], Array[@core.Field]) -> Self[S]
+pub fn[S] LibraryAsyncLogger::child(Self[S], String) -> Self[S]
+pub async fn[S] LibraryAsyncLogger::error(Self[S], String, fields? : Array[@core.Field]) -> Unit
+pub async fn[S] LibraryAsyncLogger::info(Self[S], String, fields? : Array[@core.Field]) -> Unit
+pub fn[S] LibraryAsyncLogger::is_enabled(Self[S], @core.Level) -> Bool
+pub async fn[S] LibraryAsyncLogger::log(Self[S], @core.Level, String, fields? : Array[@core.Field], target? : String) -> Unit
+pub fn[S] LibraryAsyncLogger::new(S, config? : @utils.AsyncLoggerConfig, min_level? : @core.Level, target? : String, flush? : (S) -> Unit raise) -> Self[S]
+pub async fn[S : @src.Sink] LibraryAsyncLogger::run(Self[S]) -> Unit
+pub async fn[S] LibraryAsyncLogger::shutdown(Self[S], clear? : Bool) -> Unit
+pub fn[S] LibraryAsyncLogger::to_async_logger(Self[S]) -> AsyncLogger[S]
+pub async fn[S] LibraryAsyncLogger::warn(Self[S], String, fields? : Array[@core.Field]) -> Unit
+pub fn[S] LibraryAsyncLogger::with_context_fields(Self[S], Array[@core.Field]) -> Self[S]
+pub fn[S] LibraryAsyncLogger::with_target(Self[S], String) -> Self[S]
+
+// Type aliases
+pub type ApplicationAsyncLogger = AsyncLogger[@src.RuntimeSink]
+
+pub type ApplicationTextAsyncLogger = AsyncLogger[@src.FormattedConsoleSink]
+
+pub using @utils {type AsyncFlushPolicy}
+
+pub using @utils {type AsyncLifecyclePhase}
+
+pub using @utils {type AsyncLoggerBuildConfig}
+
+pub using @utils {type AsyncLoggerConfig}
+
+pub using @utils {type AsyncLoggerState}
+
+pub using @utils {type AsyncOverflowPolicy}
+
+pub using @utils {type AsyncRuntimeMode}
+
+pub using @utils {type AsyncRuntimeState}
+
+// Traits
+

src-async/utils/async_models.mbt

@@ -50,14 +50,74 @@ pub fn AsyncRuntimeState::new(
   { mode, background_worker }
 }
 
+///|
+pub(all) enum AsyncLifecyclePhase {
+  Ready
+  Running
+  Failed
+  Closing
+  Closed
+  ClosedFailed
+}
+
+///|
+fn async_lifecycle_phase_label(phase : AsyncLifecyclePhase) -> String {
+  match phase {
+    AsyncLifecyclePhase::Ready => "ready"
+    AsyncLifecyclePhase::Running => "running"
+    AsyncLifecyclePhase::Failed => "failed"
+    AsyncLifecyclePhase::Closing => "closing"
+    AsyncLifecyclePhase::Closed => "closed"
+    AsyncLifecyclePhase::ClosedFailed => "closed_failed"
+  }
+}
+
+///|
+fn async_lifecycle_phase_is_closed(phase : AsyncLifecyclePhase) -> Bool {
+  match phase {
+    AsyncLifecyclePhase::Closing |
+    AsyncLifecyclePhase::Closed |
+    AsyncLifecyclePhase::ClosedFailed => true
+    _ => false
+  }
+}
+
+///|
+fn async_lifecycle_phase_is_running(phase : AsyncLifecyclePhase) -> Bool {
+  match phase {
+    AsyncLifecyclePhase::Running | AsyncLifecyclePhase::Closing => true
+    _ => false
+  }
+}
+
+///|
+fn async_lifecycle_phase_has_failed(phase : AsyncLifecyclePhase) -> Bool {
+  match phase {
+    AsyncLifecyclePhase::Failed | AsyncLifecyclePhase::ClosedFailed => true
+    _ => false
+  }
+}
+
+///|
+fn async_lifecycle_phase_can_rerun(phase : AsyncLifecyclePhase) -> Bool {
+  match phase {
+    AsyncLifecyclePhase::Ready | AsyncLifecyclePhase::Failed => true
+    _ => false
+  }
+}
+
 ///|
 pub struct AsyncLoggerState {
   runtime : AsyncRuntimeState
+  phase : AsyncLifecyclePhase
   pending_count : Int
   dropped_count : Int
   is_closed : Bool
   is_running : Bool
   has_failed : Bool
+  backlog_retained : Bool
+  can_rerun : Bool
+  terminal : Bool
   last_error : String
   flush_policy : AsyncFlushPolicy
 }
@@ -65,21 +125,29 @@ pub struct AsyncLoggerState {
 ///|
 pub fn AsyncLoggerState::new(
   runtime : AsyncRuntimeState,
+  phase : AsyncLifecyclePhase,
   pending_count : Int,
   dropped_count : Int,
-  is_closed : Bool,
-  is_running : Bool,
-  has_failed : Bool,
   last_error : String,
   flush_policy : AsyncFlushPolicy,
 ) -> AsyncLoggerState {
+  let is_closed = async_lifecycle_phase_is_closed(phase)
+  let is_running = async_lifecycle_phase_is_running(phase)
+  let has_failed = async_lifecycle_phase_has_failed(phase)
+  let backlog_retained = pending_count > 0 && !is_running
+  let can_rerun = async_lifecycle_phase_can_rerun(phase)
+  let terminal = is_closed && !is_running && pending_count == 0
   {
     runtime,
+    phase,
     pending_count,
     dropped_count,
     is_closed,
     is_running,
     has_failed,
+    backlog_retained,
+    can_rerun,
+    terminal,
     last_error,
     flush_policy,
   }
@@ -123,6 +191,9 @@ fn async_logger_state_to_json_value(
 ) -> @json_parser.JsonValue {
   @json_parser.JsonValue::Object({
     "runtime": async_runtime_state_to_json(state.runtime),
+    "phase": @json_parser.JsonValue::String(
+      async_lifecycle_phase_label(state.phase),
+    ),
     "pending_count": @json_parser.JsonValue::Number(
       state.pending_count.to_double(),
     ),
@@ -132,6 +203,11 @@ fn async_logger_state_to_json_value(
     "is_closed": @json_parser.JsonValue::Bool(state.is_closed),
     "is_running": @json_parser.JsonValue::Bool(state.is_running),
     "has_failed": @json_parser.JsonValue::Bool(state.has_failed),
+    "backlog_retained": @json_parser.JsonValue::Bool(
+      state.backlog_retained,
+    ),
+    "can_rerun": @json_parser.JsonValue::Bool(state.can_rerun),
+    "terminal": @json_parser.JsonValue::Bool(state.terminal),
     "last_error": @json_parser.JsonValue::String(state.last_error),
     "flush_policy": @json_parser.JsonValue::String(
       async_flush_policy_label(state.flush_policy),