feat: allow SerializableReadTransaction writing

Author: emptybutton

src/tgdb/application/horizon/commit_transaction.py

@@ -33,7 +33,7 @@ class CommitTransaction:
     clock: Clock
     relations: Relations
     channel: Channel
-    commit_buffer: Buffer[PreparedCommit]
+    commit_buffer: Buffer[Commit | PreparedCommit]
 
     async def __call__(
         self, xid: XID, operators: Sequence[Operator]
@@ -52,9 +52,6 @@ async def __call__(
         async with self.shared_horizon as horizon:
             commit = horizon.commit_transaction(time, xid, effects)
 
-        if isinstance(commit, Commit):
-            return
-
         notification, _ = await gather(
             self.channel.wait(commit.xid),
             self.commit_buffer.add(commit),

src/tgdb/application/horizon/output_commits.py

@@ -10,14 +10,14 @@
     InvalidTransactionStateError,
     NoTransactionError,
 )
-from tgdb.entities.horizon.transaction import XID, PreparedCommit
+from tgdb.entities.horizon.transaction import XID, Commit, PreparedCommit
 
 
 @dataclass(frozen=True)
 class OutputCommits:
-    commit_buffer: Buffer[PreparedCommit]
+    commit_buffer: Buffer[Commit | PreparedCommit]
     channel: Channel
-    output_commits: Queque[Sequence[PreparedCommit]]
+    output_commits: Queque[Sequence[Commit | PreparedCommit]]
     shared_horizon: SharedHorizon
     clock: Clock
 

src/tgdb/application/horizon/output_commits_to_tuples.py

@@ -3,13 +3,13 @@
 
 from tgdb.application.common.ports.queque import Queque
 from tgdb.application.relation.ports.tuples import Tuples
-from tgdb.entities.horizon.transaction import PreparedCommit
+from tgdb.entities.horizon.transaction import Commit, PreparedCommit
 
 
 @dataclass(frozen=True)
 class OutputCommitsToTuples:
     tuples: Tuples
-    output_commits: Queque[Sequence[PreparedCommit]]
+    output_commits: Queque[Sequence[Commit | PreparedCommit]]
 
     async def __call__(self) -> None:
         is_previous_map_partial = True

src/tgdb/entities/horizon/horizon.py

@@ -9,9 +9,9 @@
     Commit,
     ConflictError,
     IsolationLevel,
-    NonSerializableReadTransaction,
     NonSerializableWriteTransactionError,
     PreparedCommit,
+    ReadUncommitedTransaction,
     SerializableTransaction,
     SerializableTransactionState,
     Transaction,
@@ -58,8 +58,8 @@ class Horizon:
     _max_len: int
     _max_transaction_age: LogicTime
     _serializable_transaction_map: OrderedDict[XID, SerializableTransaction]
-    _non_serializable_read_transaction_map: OrderedDict[
-        XID, NonSerializableReadTransaction
+    _read_uncommited_transaction_map: OrderedDict[
+        XID, ReadUncommitedTransaction
     ]
 
     def __post_init__(self) -> None:
@@ -162,7 +162,8 @@ def commit_transaction(
             match transaction:
                 case SerializableTransaction():
                     return transaction.prepare_commit()
-                case NonSerializableReadTransaction():
+                case ReadUncommitedTransaction():
+                    del self._transaction_map(transaction)[xid]
                     return transaction.commit()
         except (ConflictError, NonSerializableWriteTransactionError) as error:
             del self._transaction_map(transaction)[xid]
@@ -263,12 +264,12 @@ def _serializable_transaction(
 
     def _non_serializable_read_transaction(
         self, xid: XID
-    ) -> NonSerializableReadTransaction:
+    ) -> ReadUncommitedTransaction:
         """
         :raises tgdb.entities.horizon.horizon.NoTransactionError:
         """
 
-        transaction = self._non_serializable_read_transaction_map.get(xid)
+        transaction = self._read_uncommited_transaction_map.get(xid)
 
         if transaction is None:
             raise NoTransactionError
@@ -292,14 +293,14 @@ def _transaction(
 
     def _transaction_maps(self) -> Iterable[Mapping[XID, Transaction]]:
         yield self._serializable_transaction_map
-        yield self._non_serializable_read_transaction_map
+        yield self._read_uncommited_transaction_map
 
     def _transaction_map[TransactionT: Transaction](
         self, transaction: TransactionT
     ) -> OrderedDict[XID, TransactionT]:
         match transaction:
-            case NonSerializableReadTransaction():
-                return self._non_serializable_read_transaction_map  # type: ignore[return-value]
+            case ReadUncommitedTransaction():
+                return self._read_uncommited_transaction_map  # type: ignore[return-value]
             case SerializableTransaction():
                 return self._serializable_transaction_map  # type: ignore[return-value]
 
@@ -316,5 +317,5 @@ def horizon(
         _max_len=max_len,
         _max_transaction_age=max_transaction_age,
         _serializable_transaction_map=OrderedDict(),
-        _non_serializable_read_transaction_map=OrderedDict(),
+        _read_uncommited_transaction_map=OrderedDict(),
     )

src/tgdb/entities/horizon/transaction.py

@@ -233,11 +233,10 @@ def _complete(self) -> None:
 
 
 @dataclass
-class NonSerializableReadTransaction:
+class ReadUncommitedTransaction:
     _xid: XID
     _start_time: LogicTime
-    _is_readonly: bool
-    _is_completed: bool
+    _space: dict[TID, TupleEffect]
 
     def xid(self) -> XID:
         return self._xid
@@ -249,54 +248,46 @@ def age(self, time: LogicTime) -> LogicTime:
         return time - self._start_time
 
     def include(self, effect: ConflictableTransactionScalarEffect) -> None:
-        if self._is_readonly and not isinstance(
-            effect, JustViewedTuple | MigratedTuple
-        ):
-            self._is_readonly = False
-
-    def rollback(self) -> None:
-        self._is_completed = True
+        if isinstance(effect, JustViewedTuple | Claim):
+            return
 
-    def commit(self) -> Commit:
-        """
-        :raises tgdb.entities.horizon.transaction.NonSerializableWriteTransactionError:
-        """  # noqa: E501
+        if effect.tid in self._space:
+            effect = self._space[effect.tid] & effect
 
-        self._is_completed = True
+        self._space[effect.tid] = effect
 
-        if not self._is_readonly:
-            self.rollback()
-            raise NonSerializableWriteTransactionError(self._xid)
+    def rollback(self) -> None:
+        ...
 
+    def commit(self) -> Commit:
         return Commit(self._xid, frozenset())
 
     @classmethod
     def start(
         cls, xid: XID, time: LogicTime
-    ) -> "NonSerializableReadTransaction":
-        return NonSerializableReadTransaction(
+    ) -> "ReadUncommitedTransaction":
+        return ReadUncommitedTransaction(
             _xid=xid,
             _start_time=time,
-            _is_readonly=True,
-            _is_completed=False,
+            _space=dict(),
         )
 
     def __eq__(self, other: object) -> bool:
         return (
-            isinstance(other, SerializableTransaction)
+            isinstance(other, ReadUncommitedTransaction)
             and self.xid() == other.xid()
         )
 
     def __hash__(self) -> int:
         return hash(type(self)) + hash(self._xid)
 
 
-type Transaction = SerializableTransaction | NonSerializableReadTransaction
+type Transaction = SerializableTransaction | ReadUncommitedTransaction
 
 
 class IsolationLevel(Enum):
-    serializable_read_and_write = auto()
-    non_serializable_read = auto()
+    serializable = auto()
+    read_uncommited = auto()
 
 
 def start_transaction(
@@ -306,10 +297,10 @@ def start_transaction(
     serializable_transactions: Iterable[SerializableTransaction],
 ) -> Transaction:
     match isolation:
-        case IsolationLevel.serializable_read_and_write:
+        case IsolationLevel.serializable:
             return SerializableTransaction.start(
                 xid, time, serializable_transactions
             )
 
-        case IsolationLevel.non_serializable_read:
-            return NonSerializableReadTransaction.start(xid, time)
+        case IsolationLevel.read_uncommited:
+            return ReadUncommitedTransaction.start(xid, time)

src/tgdb/infrastructure/adapters/buffer.py

@@ -7,7 +7,7 @@
 from typing import Self
 
 from tgdb.application.common.ports.buffer import Buffer
-from tgdb.entities.horizon.transaction import PreparedCommit
+from tgdb.entities.horizon.transaction import Commit, PreparedCommit
 from tgdb.infrastructure.telethon.in_telegram_bytes import InTelegramBytes
 
 
@@ -47,8 +47,8 @@ def _refresh_overflow(self) -> None:
 
 
 @dataclass(frozen=True)
-class InTelegramReplicablePreparedCommitBuffer(Buffer[PreparedCommit]):
-    _buffer: Buffer[PreparedCommit]
+class InTelegramReplicablePreparedCommitBuffer(Buffer[Commit | PreparedCommit]):
+    _buffer: Buffer[Commit | PreparedCommit]
     _in_tg_encoded_commits: InTelegramBytes
 
     async def __aenter__(self) -> Self:
@@ -77,10 +77,12 @@ async def __aexit__(
         traceback: TracebackType | None,
     ) -> None: ...
 
-    async def add(self, commit: PreparedCommit, /) -> None:
+    async def add(self, commit: Commit | PreparedCommit, /) -> None:
         await self._buffer.add(commit)
 
-    async def __aiter__(self) -> AsyncIterator[Sequence[PreparedCommit]]:
+    async def __aiter__(
+        self
+    ) -> AsyncIterator[Sequence[Commit | PreparedCommit]]:
         async for commits in self._buffer:
             encoded_commits = pickle.dumps(list(commits))
             await self._in_tg_encoded_commits.set(encoded_commits)

src/tgdb/infrastructure/telethon/lazy_map.py

@@ -3,9 +3,8 @@
 from telethon.hints import TotalList
 from telethon.tl.types import Message
 
-from tgdb.entities.relation.tuple import TID
 from tgdb.infrastructure.heap_tuple_encoding import HeapTupleEncoding
-from tgdb.infrastructure.lazy_map import ExternalValue, LazyMap, NoExternalValue
+from tgdb.infrastructure.lazy_map import LazyMap
 from tgdb.infrastructure.telethon.client_pool import TelegramClientPool
 from tgdb.infrastructure.telethon.index import (
     MessageIndex,

src/tgdb/main/common/di.py

@@ -23,7 +23,7 @@
 from tgdb.application.relation.ports.tuples import Tuples
 from tgdb.application.relation.view_tuples import ViewTuples
 from tgdb.entities.horizon.horizon import Horizon, horizon
-from tgdb.entities.horizon.transaction import PreparedCommit
+from tgdb.entities.horizon.transaction import Commit, PreparedCommit
 from tgdb.entities.relation.relation import Relation
 from tgdb.infrastructure.adapters.buffer import (
     InMemoryBuffer,
@@ -71,7 +71,7 @@ class CommonProvider(Provider):
     provide_uuids = provide(UUIDs4, provides=UUIDs, scope=Scope.APP)
     provide_queque = provide(
         staticmethod(lambda: InMemoryQueque(AsyncQueque())),
-        provides=Queque[Sequence[PreparedCommit]],
+        provides=Queque[Sequence[Commit | PreparedCommit]],
         scope=Scope.APP
     )
     provide_channel = provide(
@@ -163,8 +163,8 @@ async def provide_buffer(
         conf: Conf,
         bot_pool: BotPool,
         user_bot_pool: UserBotPool,
-        buffer: InMemoryBuffer[PreparedCommit]
-    ) -> AsyncIterator[Buffer[PreparedCommit]]:
+        buffer: InMemoryBuffer[Commit | PreparedCommit]
+    ) -> AsyncIterator[Buffer[Commit | PreparedCommit]]:
         in_tg_bytes = InTelegramBytes(bot_pool, user_bot_pool, conf.buffer.chat)
 
         biffer = InTelegramReplicablePreparedCommitBuffer(buffer, in_tg_bytes)

src/tgdb/presentation/fastapi/horizon/schemas/isolation_level.py

@@ -4,26 +4,24 @@
 from tgdb.entities.horizon.transaction import IsolationLevel
 
 
-type _EncodedIsolationLevel = Literal[
-    "serializableReadAndWrite", "nonSerializableRead"
-]
+type _EncodedIsolationLevel = Literal["serializable", "readUncommited"]
 
 
 class IsolationLevelSchema(StrEnum):
-    serializable_read_and_write = "serializableReadAndWrite"
-    non_serializable_read = "nonSerializableRead"
+    serializable = "serializable"
+    read_uncommited = "readUncommited"
 
     def decoded(self) -> IsolationLevel:
         match self:
-            case IsolationLevelSchema.serializable_read_and_write:
-                return IsolationLevel.serializable_read_and_write
-            case IsolationLevelSchema.non_serializable_read:
-                return IsolationLevel.non_serializable_read
+            case IsolationLevelSchema.serializable:
+                return IsolationLevel.serializable
+            case IsolationLevelSchema.read_uncommited:
+                return IsolationLevel.read_uncommited
 
     @classmethod
     def of(cls, level: IsolationLevel) -> "IsolationLevelSchema":
         match level:
-            case IsolationLevel.serializable_read_and_write:
-                return IsolationLevelSchema.serializable_read_and_write
-            case IsolationLevel.non_serializable_read:
-                return IsolationLevelSchema.non_serializable_read
+            case IsolationLevel.serializable:
+                return IsolationLevelSchema.serializable
+            case IsolationLevel.read_uncommited:
+                return IsolationLevelSchema.read_uncommited