bitflicker64 commented on code in PR #391:
URL: https://github.com/apache/hugegraph-ai/pull/391#discussion_r4418791457
##########
ontogeny/server/ontogeny/auth/service.py:
##########
@@ -284,37 +326,33 @@ async def ensure_admin(self) -> dict[str, Any] | None:
return None
username = os.environ.get("ONTOGENY_ADMIN_USER", "admin")
- # the documented dev bootstrap default: a random value that is never
- # retrievable afterwards would lock every fresh non-env deployment out
- password = os.environ.get("ONTOGENY_ADMIN_PASSWORD") or
BOOTSTRAP_ADMIN_PASSWORD
- # self-heal instead of crashing: the bootstrap account may already
- # exist as non-admin or disabled (demoted by a buggy older path), and
- # create_user would raise "username exists" on the next restart
+ # Never silently restore privileges to a disabled or demoted account.
+ # A deployment with this inconsistent state needs explicit recovery.
async with self.sc.sessionmaker() as session:
row = await self._user(session, username)
if row is not None:
- row.is_admin = True
- row.status = "active"
- row.roles = sorted(set(row.roles) | {"admin"})
- await session.commit()
- log.warning(
- "re-activated the bootstrap administrator %r (it existed
but was "
- "not an active admin); set ONTOGENY_ADMIN_PASSWORD and
change it",
- username,
+ raise ConflictError(
+ f"no active administrator exists and bootstrap user
{username!r} already exists; "
+ "restore an administrator explicitly or select a new
ONTOGENY_ADMIN_USER"
)
- return self._view(row)
+ configured_password = os.environ.get("ONTOGENY_ADMIN_PASSWORD")
+ password = configured_password or f"{secrets.randbelow(1_000_000):06d}"
Review Comment:
Important: the generated admin password has 10^6 values, the username
defaults to `admin`, and `/auth/login` has no attempt limit, while
`MIN_PASSWORD = 6` lowers the floor for every account. Generate a high-entropy
bootstrap secret such as `secrets.token_urlsafe(24)` and keep `MIN_PASSWORD =
8`.
##########
ontogeny/server/ontogeny/action/outbox.py:
##########
@@ -155,31 +157,82 @@ async def _deliver_sse(self, event: OutboxRow) -> bool:
)
return True
- async def _deliver_derivation(self, event: OutboxRow) -> bool:
+ async def _deliver_derivation(self, event: OutboxRow, session) -> bool:
if self.derivation_hook is None:
return True
- return bool(await self.derivation_hook(event))
+ return bool(await self.derivation_hook(event, session))
# --------------------------------------------------------------- driving
async def dispatch_pending(self, *, limit: int = 100) -> dict[str, int]:
- consumers: dict[str, Consumer] = {
- "webhook": self._deliver_webhook,
- "projection": self._deliver_projection,
- "derivation": self._deliver_derivation,
- "sse": self._deliver_sse,
- }
+ # The background loop and the admin endpoint share cursors and must
+ # never dispatch the same batch concurrently in this process.
+ async with self._dispatch_lock:
+ return await self._dispatch_pending(limit=limit)
+
+ async def replay(self, consumer: str, *, from_id: int, through_id: int) ->
int:
+ """Explicitly retry acknowledged events in a bounded inclusive range.
+
+ Replaying a webhook may deliver it again. This also permits operators
+ to recover old pre-migration gaps that legacy cursors cannot
distinguish
+ from completed deliveries.
+ """
+ if consumer not in {"webhook", "projection", "derivation", "sse"} or
from_id < 1 or through_id < from_id:
+ raise ValueError("invalid outbox consumer or replay range")
+ async with self._dispatch_lock, self.session_factory() as session:
+ cursor = await session.get(OutboxConsumerRow, consumer)
+ if cursor is None or not cursor.receipts_initialized:
+ raise ValueError("consumer receipts have not been initialized;
dispatch first")
+ result = await session.execute(
+ delete(OutboxReceiptRow).where(
+ OutboxReceiptRow.consumer == consumer,
+ OutboxReceiptRow.event_id >= from_id,
+ OutboxReceiptRow.event_id <= through_id,
+ )
+ )
+ await session.commit()
+ return int(result.rowcount)
+
+ async def _dispatch_pending(self, *, limit: int = 100) -> dict[str, int]:
stats: dict[str, int] = {}
async with self.session_factory() as session:
+ consumers: dict[str, Consumer] = {
+ "webhook": self._deliver_webhook,
+ "projection": self._deliver_projection,
+ "derivation": lambda event: self._deliver_derivation(event,
session),
+ "sse": self._deliver_sse,
+ }
for name, deliver in consumers.items():
cursor = (
await
session.execute(select(OutboxConsumerRow).where(OutboxConsumerRow.consumer ==
name))
).scalar_one_or_none()
- last_id = cursor.last_id if cursor else 0
+ if cursor is None:
+ cursor = OutboxConsumerRow(consumer=name, last_id=0,
receipts_initialized=True)
+ session.add(cursor)
+ await session.flush()
+ elif not cursor.receipts_initialized:
+ # Snapshot the *visible* legacy history once. Do not replay
+ # old webhooks implicitly; a transaction that commits after
+ # this statement has no receipt even if its ID is lower.
+ await session.execute(
+ insert(OutboxReceiptRow).from_select(
+ ["consumer", "event_id"],
+ select(literal(name),
OutboxRow.id).where(OutboxRow.id <= cursor.last_id),
+ )
+ )
+ cursor.receipts_initialized = True
+ await session.flush()
rows = (
(
await session.execute(
- select(OutboxRow).where(OutboxRow.id >
last_id).order_by(OutboxRow.id).limit(limit)
+ select(OutboxRow)
+ .where(
+ ~exists().where(
Review Comment:
Important: this anti-join starts at the first outbox row and nothing prunes
`ontogeny_outbox` or `ontogeny_outbox_receipt`, so each 1-second tick runs four
scans whose cost grows with all delivered history (every sync adds one row per
source row). A local SQLite run with 1M delivered events spends about 1 s per
idle tick. Keep a per-consumer floor below which every ID is acknowledged and
add `OutboxRow.id > floor` to the query.
##########
ontogeny/server/ontogeny/functions/derivation.py:
##########
@@ -149,16 +207,33 @@ async def _compute(self, object_type: str, prop: str,
entry: str, ids: list[str]
values = await self.sandbox.map_rows(s, fn, assembled, {},
object_param=object_param)
except Exception as exc:
log.warning("derivation %s.%s failed for %d rows: %s",
object_type, prop, len(rows), exc)
- return 0
+ return 0, set()
Review Comment:
Important: a row whose function raises fails the whole `map_rows` batch, and
its pending entry now persists, so every later drain includes it and no row of
that type gets new derived values while it stays pending. Retry a failed batch
row by row and leave only the rows that still fail pending.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
