feat: make it better, faster, stronger

Author: emptybutton

src/tgdb/application/input_operator.py

@@ -1,15 +1,23 @@
 from dataclasses import dataclass
 
-from tgdb.application.errors.common import InvalidInputOperatorError
+from tgdb.application.errors.common import (
+    InvalidInputOperatorError,
+    NotActiveNodeError,
+)
+from tgdb.application.ports.buffer import Buffer
+from tgdb.application.ports.logic_clock import LogicClock
 from tgdb.application.ports.operator_serialization import OperatorSerialization
-from tgdb.application.ports.queque import Queque
-from tgdb.entities.operator import Operator
+from tgdb.application.ports.shared_horizon import SharedHorizon
+from tgdb.entities.operator import AppliedOperator
+from tgdb.entities.transaction import TransactionPreparedCommit
 
 
 @dataclass(frozen=True)
 class InputOperator[SerializedOperatorsT]:
-    input_operators: Queque[Operator]
+    clock: LogicClock
     operator_serialization: OperatorSerialization[SerializedOperatorsT]
+    commit_buffer: Buffer[TransactionPreparedCommit]
+    shared_horizon: SharedHorizon
 
     async def __call__(
         self, serialized_operator: SerializedOperatorsT
@@ -25,4 +33,11 @@ async def __call__(
         if input_operator is None:
             raise InvalidInputOperatorError
 
-        await self.input_operators.push(input_operator)
+        async with self.shared_horizon as horizon:
+            time = await self.clock.time()
+            applied_input_operator = AppliedOperator(input_operator, time)
+
+            transaction_commit = horizon.add(applied_input_operator)
+
+            if transaction_commit:
+                await self.commit_buffer.add(transaction_commit)

src/tgdb/application/output_commits.py

@@ -2,16 +2,34 @@
 from dataclasses import dataclass
 
 from tgdb.application.ports.buffer import Buffer
+from tgdb.application.ports.notifying import Notifying
 from tgdb.application.ports.queque import Queque
-from tgdb.entities.transaction import TransactionCommit
+from tgdb.application.ports.shared_horizon import SharedHorizon
+from tgdb.entities.transaction import (
+    TransactionOkPreparedCommit,
+    TransactionPreparedCommit,
+)
 
 
 @dataclass(frozen=True)
 class OutputCommits:
-    commit_buffer: Buffer[TransactionCommit]
-    output_commits: Queque[Sequence[TransactionCommit]]
+    commit_buffer: Buffer[TransactionPreparedCommit]
+    notifying: Notifying[Sequence[TransactionPreparedCommit]]
+    output_commits: Queque[Sequence[TransactionOkPreparedCommit]]
+    shared_horizon: SharedHorizon
 
     async def __call__(self) -> None:
-        async for commits in self.commit_buffer:
-            await self.output_commits.push(commits)
+        async for prepared_commits in self.commit_buffer:
+            await self.notifying.publish(prepared_commits)
+
+            ok_prepared_commits = tuple(
+                commit for commit in prepared_commits
+                if isinstance(commit, TransactionOkPreparedCommit)
+            )
+
+            await self.output_commits.push(ok_prepared_commits)
             await self.output_commits.sync()
+
+            async with self.shared_horizon as horizon:
+                for ok_prepared_commit in ok_prepared_commits:
+                    horizon.complete(ok_prepared_commit)

src/tgdb/application/output_commits_to_clients.py

@@ -3,26 +3,16 @@
 
 from tgdb.application.ports.heap import Heap
 from tgdb.application.ports.queque import Queque
-from tgdb.entities.transaction import TransactionCommit, TransactionOkCommit
+from tgdb.entities.transaction import TransactionOkPreparedCommit
 
 
 @dataclass(frozen=True)
 class OutputCommitsToHeap:
     heap: Heap
-    output_commits: Queque[Sequence[TransactionCommit]]
+    output_commits: Queque[Sequence[TransactionOkPreparedCommit]]
 
     async def __call__(self) -> None:
-        is_previous_map_partial = True
-
         async for commits in self.output_commits:
-            effects = tuple(
-                commit.effect
-                for commit in commits
-                if isinstance(commit, TransactionOkCommit)
-            )
+            effects = tuple(commit.effect for commit in commits)
 
-            if is_previous_map_partial:
-                await self.heap.map_as_duplicate(effects)
-                is_previous_map_partial = False
-            else:
-                await self.heap.map(effects)
+            await self.heap.map(effects)

src/tgdb/application/output_commits_to_heap.py

@@ -10,8 +10,3 @@ async def rows(self, schema: str, )
 
     @abstractmethod
     async def map(self, effects: Sequence[TransactionEffect], /) -> None: ...
-
-    @abstractmethod
-    async def map_as_duplicate(
-        self, effects: Sequence[TransactionEffect], /
-    ) -> None: ...

src/tgdb/application/ports/heap.py

@@ -0,0 +1,6 @@
+from abc import ABC, abstractmethod
+
+
+class Notifying[ValueT](ABC):
+    @abstractmethod
+    async def publish(self, value: ValueT, /) -> None: ...

src/tgdb/application/ports/notification.py

@@ -4,8 +4,4 @@
 from tgdb.entities.transaction_horizon import TransactionHorizon
 
 
-class SharedHorizon(
-    AbstractAsyncContextManager[TransactionHorizon | None], ABC
-):
-    @abstractmethod
-    async def set(self, horizon: TransactionHorizon | None) -> None: ...
+class SharedHorizon(AbstractAsyncContextManager[TransactionHorizon], ABC): ...