Spaces:
Running
Running
File size: 8,964 Bytes
24480a0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 | """
backend/memory/oracle_sync.py — Supabase ↔ Oracle DB dual-write synchronizer.
Two public classes:
MemorySynchronizer — Wraps a primary (Supabase) and a secondary (Oracle)
MemoryBackend. Writes always go to both; Oracle failures are non-fatal
(logged but not re-raised) because Oracle is a secondary replica, not
the source of truth.
OracleState — Singleton lifecycle manager hooked from main.py _on_startup().
Reads ORACLE_DB_* env vars, initialises OracleMemoryAdapter if they are
set, and exposes oracle_adapter and memory_synchronizer as class-level
attributes for use by other API modules.
Design decisions:
- IDs are generated once by MemorySynchronizer (uuid4) and passed to BOTH
backends so keys stay consistent (BUG-1 fix).
- Secondary write is wrapped in try/except so a primary success is never
rolled back due to an Oracle failure (BUG-4 fix).
- sync_update / sync_delete return a dict {'primary': bool, 'secondary': bool|None}
so callers can distinguish which backend failed (BUG-8 fix).
- OracleState checks oracle_adapter.is_connected, not the object's truthiness,
to avoid initialising the synchronizer against an unconnected adapter (BUG-3 fix).
"""
from __future__ import annotations
import logging
import os
import uuid
from typing import Any, Dict, List, Optional
from .memory_backend import MemoryBackend
from .oracle_adapter import OracleMemoryAdapter
_logger = logging.getLogger("memory.oracle_sync")
# ── Synchronizer ─────────────────────────────────────────────────────────────
class MemorySynchronizer:
"""
Dual-write coordinator: primary backend is authoritative, secondary is
best-effort.
Args:
primary: Authoritative backend (Supabase). Errors propagate.
secondary: Replica backend (Oracle). Errors are logged, never raised.
"""
def __init__(self, primary: MemoryBackend, secondary: OracleMemoryAdapter) -> None:
self.primary = primary
self.secondary = secondary
# ── Writes ───────────────────────────────────────────────────────────────
async def sync_add_memory(
self, user_id: str, memory_data: Dict[str, Any]
) -> str:
"""Insert into primary then Oracle. Returns the shared memory_id."""
shared_id = str(uuid.uuid4()) # single ID for both backends
await self.primary.add_memory(user_id, memory_data, memory_id=shared_id)
if self.secondary.is_connected:
try:
await self.secondary.add_memory(
user_id, memory_data, memory_id=shared_id
)
except Exception as exc:
_logger.warning(
"OracleSync.add_memory: Oracle write failed (non-fatal) — %s", exc
)
return shared_id
async def sync_update_memory(
self, user_id: str, memory_id: str, new_data: Dict[str, Any]
) -> Dict[str, Optional[bool]]:
"""
Update both backends.
Returns:
{'primary': bool, 'secondary': bool | None}
None means Oracle is not connected or the call was skipped.
"""
primary_ok = await self.primary.update_memory(user_id, memory_id, new_data)
secondary_ok: Optional[bool] = None
if self.secondary.is_connected:
try:
secondary_ok = await self.secondary.update_memory(
user_id, memory_id, new_data
)
except Exception as exc:
_logger.warning(
"OracleSync.update_memory: Oracle update failed — %s", exc
)
secondary_ok = False
return {"primary": primary_ok, "secondary": secondary_ok}
async def sync_delete_memory(
self, user_id: str, memory_id: str
) -> Dict[str, Optional[bool]]:
"""
Delete from both backends.
Returns:
{'primary': bool, 'secondary': bool | None}
"""
primary_ok = await self.primary.delete_memory(user_id, memory_id)
secondary_ok: Optional[bool] = None
if self.secondary.is_connected:
try:
secondary_ok = await self.secondary.delete_memory(user_id, memory_id)
except Exception as exc:
_logger.warning(
"OracleSync.delete_memory: Oracle delete failed — %s", exc
)
secondary_ok = False
return {"primary": primary_ok, "secondary": secondary_ok}
# ── Reads ────────────────────────────────────────────────────────────────
async def get_memory_from_primary(
self, user_id: str, memory_id: str
) -> Optional[Dict[str, Any]]:
return await self.primary.get_memory(user_id, memory_id)
async def get_memory_from_secondary(
self, user_id: str, memory_id: str
) -> Optional[Dict[str, Any]]:
if not self.secondary.is_connected:
return None
return await self.secondary.get_memory(user_id, memory_id)
async def list_memories_from_primary(
self, user_id: str, limit: int = 100, offset: int = 0
) -> List[Dict[str, Any]]:
return await self.primary.list_memories(user_id, limit, offset)
async def list_memories_from_secondary(
self, user_id: str, limit: int = 100, offset: int = 0
) -> List[Dict[str, Any]]:
if not self.secondary.is_connected:
return []
return await self.secondary.list_memories(user_id, limit, offset)
# ── Lifecycle singleton ───────────────────────────────────────────────────────
class OracleState:
"""
Singleton lifecycle manager for the Oracle memory backend.
Hooked from backend/main.py::_on_startup() — call await OracleState.initialize().
The synchronizer is wired lazily; if ORACLE_DB_* vars are absent the class
silently stays disabled so the rest of the API is unaffected.
Usage after startup:
from memory.oracle_sync import OracleState
adapter = OracleState.oracle_adapter # None if not configured
sync = OracleState.memory_synchronizer # None if not configured
"""
oracle_adapter: Optional[OracleMemoryAdapter] = None
memory_synchronizer: Optional[MemorySynchronizer] = None
_initialized: bool = False
@classmethod
async def initialize(cls) -> None:
if cls._initialized:
return
user = os.getenv("ORACLE_DB_USER")
password = os.getenv("ORACLE_DB_PASSWORD")
dsn = os.getenv("ORACLE_DB_DSN")
if not all([user, password, dsn]):
_logger.info(
"OracleState: ORACLE_DB_* env vars not set — Oracle backend disabled."
)
cls._initialized = True
return
adapter = OracleMemoryAdapter()
try:
await adapter.connect({"user": user, "password": password, "dsn": dsn})
except Exception as exc:
_logger.warning(
"OracleState: connect failed (%s) — Oracle backend disabled.", exc
)
cls._initialized = True
return
cls.oracle_adapter = adapter
# Wire up MemorySynchronizer only when a primary backend is available.
# SupabaseMemoryBackend (backend/memory/supabase_backend.py) wraps the
# existing state._sb client. If that module doesn't exist yet the
# synchronizer stays None; oracle_adapter is still available for direct use.
try:
from memory.supabase_backend import SupabaseMemoryBackend # type: ignore[import]
cls.memory_synchronizer = MemorySynchronizer(
primary=SupabaseMemoryBackend(), secondary=cls.oracle_adapter
)
_logger.info(
"OracleState: MemorySynchronizer ready (Supabase→Oracle dual-write active)."
)
except ImportError:
_logger.info(
"OracleState: supabase_backend not found — "
"oracle_adapter available for direct use; synchronizer disabled."
)
cls._initialized = True
@classmethod
async def shutdown(cls) -> None:
"""Call from _on_shutdown() to close the Oracle pool gracefully."""
if cls.oracle_adapter and cls.oracle_adapter.is_connected:
await cls.oracle_adapter.disconnect()
cls._initialized = False
|