feat: log flush interface

Author: brian14708

src/duron/log/storage/base.py

@@ -28,3 +28,5 @@ async def acquire_lease(self) -> _TLease: ...
     async def release_lease(self, token: _TLease): ...
 
     async def append(self, token: _TLease, entry: Entry) -> _TOffset: ...
+
+    async def flush(self, token: _TLease, offset: _TOffset): ...

src/duron/log/storage/simple.py

@@ -104,6 +104,10 @@ async def append(self, token: str, entry: Entry) -> int:
 
             return entry_offset
 
+    @override
+    async def flush(self, token: str, offset: int):
+        pass
+
 
 class MemoryLogStorage(LogStorage[int, str]):
     _entries: list[AnyEntry]
@@ -176,6 +180,10 @@ async def append(self, token: str, entry: Entry) -> int:
 
             return index
 
+    @override
+    async def flush(self, token: str, offset: int):
+        pass
+
     async def entries(self) -> list[AnyEntry]:
         async with self._lock:
             return self._entries.copy()

src/duron/task.py

@@ -256,12 +256,14 @@ async def handle_message(self, e: Entry) -> None:
         else:
             pass
 
-    async def enqueue_log(self, entry: Entry) -> None:
+    async def enqueue_log(self, entry: Entry, flush: bool = False) -> None:
         if not self._running:
             self._pending_msg.append(entry)
         else:
-            _ = await self._log.append(self._running, entry)
+            offset = await self._log.append(self._running, entry)
             await self.handle_message(entry)
+            if flush:
+                await self._log.flush(self._running, offset)
 
     async def enqueue_op(self, id: bytes, op: Op | object) -> None:
         match op:
@@ -313,7 +315,7 @@ async def cb() -> None:
                     except BaseException as e:
                         entry["error"] = _encode_error(e, self._codec)
 
-                    await self.enqueue_log(entry)
+                    await self.enqueue_log(entry, True)
 
                 _ = self._loop.create_task(cb())