Merge branch 'fix/zone-sync-limits' into 'develop'

| `FM_PARTNER_STATUS_LINK` | — | Callback URL advertised to partners |

| `FM_PLATFORM_CAPS` | `["serviceAPIs"]` | Capabilities advertised on federation setup |

| `FM_BOOTSTRAP_FEDERATION` | `false` | Federate with every active partner at startup (ADR-0043) |

| `FM_ZONE_OFFER_SWEEP_SECONDS` | `600` | How often each federated partner's zone offer is re-checked and missed `ZONES` callbacks resent; agreement writes and `valid_from`/`valid_until` trigger a check on their own |

| `FM_EVENT_CONSUMER_DURABLE` | `fm-event-worker` | JetStream durable name; must differ per deployment |

| `FM_ALLOW_INSECURE_PARTNER_ENDPOINTS` | `false` | Permit plain-HTTP partner endpoints. **Development only** |

| `FM_PARTNER_SECRET_KEY` | unset | Base64url 32-byte key that encrypts partner secrets sent as `our_client_secret`. Unset: such requests get 501 and partners use `our_client_secret_ref` files. Changing it makes stored secrets unreadable |

from datetime import datetime

from typing import Any

from uuid import UUID

        )

        return [_to_agreement(row) for row in (await self._session.execute(stmt)).all()]

    async def next_validity_boundary(self, after: datetime) -> datetime | None:

        live = agreements.c.status == "active"

        starts = select(func.min(agreements.c.valid_from)).where(

            live, agreements.c.valid_from > after

        )

        ends = select(func.min(agreements.c.valid_until)).where(

            live, agreements.c.valid_until > after

        )

        found = [

            moment

            for moment in (

                (await self._session.execute(starts)).scalar_one_or_none(),

                (await self._session.execute(ends)).scalar_one_or_none(),

            )

            if moment is not None

        ]

        return min(found, default=None)

    async def add(self, agreement: Agreement) -> None:

        await self._write(

            agreement,

from typing import Any

from uuid import UUID

from sqlalchemy import Select, insert, select, update

            .limit(1)

        )

    async def list_active_inbound(self) -> list[FederationContext]:

        stmt = (

            select(contexts)

            .where(contexts.c.direction == "inbound", contexts.c.status == "available")

            .order_by(contexts.c.created_at)

        )

        return [_to_context(row) for row in (await self._session.execute(stmt)).all()]

    async def find_inbound(

        self, partner_id: UUID, federation_context_id: str

    ) -> FederationContext | None:

                direction=context.direction,

                federation_context_id=context.federation_context_id,

                status_callback_url=context.status_callback_url,

                offered_zone_ids=_zone_ids(context.offered_zone_ids),

                status=context.status,

                created_at=context.created_at,

            )

        )

        await self._session.commit()

    async def set_offered_zones(self, context_id: UUID, zone_ids: frozenset[UUID]) -> None:

        await self._session.execute(

            update(contexts)

            .where(contexts.c.id == context_id)

            .values(offered_zone_ids=_zone_ids(zone_ids))

        )

        await self._session.commit()

    async def _one(self, stmt: Select[tuple[object, ...]]) -> FederationContext | None:

        row = (await self._session.execute(stmt)).one_or_none()

        if row is None:

            return None

        return None if row is None else _to_context(row)

def _zone_ids(zone_ids: frozenset[UUID] | None) -> list[str] | None:

    return None if zone_ids is None else sorted(str(z) for z in zone_ids)

def _to_context(row: Any) -> FederationContext:

    return FederationContext(

        id=row.id,

        partner_op_id=row.partner_op_id,

        created_at=row.created_at,

        agreement_id=row.agreement_id,

        status_callback_url=row.status_callback_url,

        offered_zone_ids=(

            None

            if row.offered_zone_ids is None

            else frozenset(UUID(str(z)) for z in row.offered_zone_ids)

        ),

    )

import hashlib

from collections.abc import AsyncIterator

from contextlib import asynccontextmanager

from uuid import UUID

from sqlalchemy import text

from sqlalchemy.ext.asyncio import AsyncEngine

class PostgresPartnerLock:

    """A Postgres advisory lock per partner, so two FM replicas never send the same change."""

    def __init__(self, engine: AsyncEngine) -> None:

        self._engine = engine

    @asynccontextmanager

    async def hold(self, partner_id: UUID) -> AsyncIterator[None]:

        key = _key(partner_id)

        # a connection of its own: the lock lives as long as the connection holds it, whatever

        # the work inside commits or rolls back

        async with self._engine.connect() as connection:

            await connection.execute(text("SELECT pg_advisory_lock(:key)"), {"key": key})

            try:

                yield

            finally:

                await connection.execute(text("SELECT pg_advisory_unlock(:key)"), {"key": key})

def _key(partner_id: UUID) -> int:

    digest = hashlib.blake2b(b"fm-zone-offer:" + partner_id.bytes, digest_size=8).digest()

    return int.from_bytes(digest, "big", signed=True)

        blob: bytes | None = (await self._session.execute(stmt)).scalar_one_or_none()

        return blob

    async def set_supported_apis(self, partner_id: UUID, apis: tuple[str, ...]) -> None:

        # written only from what the partner tells us, never through the registry

        await self._session.execute(

            update(partner_ops)

            .where(partner_ops.c.id == partner_id)

            .values(supported_apis=list(apis))

        )

        await self._session.commit()

    async def add(self, partner: PartnerOP, encrypted_secret: bytes | None = None) -> None:

        await self._write(

            insert(partner_ops).values(id=partner.id, **self._values(partner)),

            base_url=row.base_url,

            name=row.name,

            our_client_secret_set=row.our_client_secret_set,

            supported_apis=tuple(row.supported_apis or ()),

        )