Commit 27cda76a authored by Sergio Gimenez's avatar Sergio Gimenez
Browse files

fix(fm): push zone changes on validity and retry, release withdrawn subscriptions 

Three gaps left by zone sync. A partner was told about zone changes only
when an agreement was written, so an agreement reaching valid_until kept
its zones on the partner's side until the partner read
GetFederationDetails, and a callback the partner didn't take was never
sent again. A withdrawn zone left both sides' subscription records behind.
And the partner's platformCaps lived only on the zones it offered, so zones
it added after withdrawing all of them carried none.

As the provider, FM now records on the inbound federation the zones the
partner was last told it has (offered_zone_ids, set by the CreateFederation
answer, GetFederationDetails and every delivered callback) and sends the
difference against what the agreement offers now. Outside valid_from and
valid_until an agreement offers nothing. The check runs on every agreement
write, at the next agreement boundary, and every FM_ZONE_OFFER_SWEEP_SECONDS
(default 600), which also resends what didn't arrive. A Postgres advisory
lock per partner keeps replicas from sending the same change. A zone that
leaves the offer drops the partner's subscription to it.

As the consumer, a zone the partner stops offering (ZONES REMOVE or a
GetFederationDetails without it) drops our accepted-zone record, and the
partner's platformCaps are kept on partner_ops.supported_apis.

Migration 0009 adds both columns and fills supported_apis from the zones
already held. Federations from before it get offered_zone_ids on FM's
first check, taken as what the partner holds.

Co-Authored-By: default avatarClaude Opus 5.5 <noreply@anthropic.com>
parent 91dac7a6

README.md

+1 −0
Changes for README.md: 1 added line, 0 removed lines.
Original line number Diff line number Diff line
@@ -98,6 +98,7 @@ Settings are environment variables with the `FM_` prefix, read by `federation_ma 
| `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 |

src/federation_manager/adapters/database/agreement_repo.py

+19 −0
Changes for src/federation_manager/adapters/database/agreement_repo.py: 19 added lines, 0 removed lines.
Original line number Diff line number Diff line 
from datetime import datetime

from typing import Any

from uuid import UUID
@@ -46,6 +47,24 @@ class PostgresAgreementRepo: 
        )

        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,

src/federation_manager/adapters/database/federation_context_repo.py

+31 −2
Changes for src/federation_manager/adapters/database/federation_context_repo.py: 31 added lines, 2 removed lines.
Original line number Diff line number Diff line 
from typing import Any

from uuid import UUID



from sqlalchemy import Select, insert, select, update
@@ -35,6 +36,14 @@ class PostgresFederationContextRepo: 
            .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:
@@ -69,6 +78,7 @@ class PostgresFederationContextRepo: 
                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,

            )
@@ -81,10 +91,24 @@ class PostgresFederationContextRepo: 
        )

        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,
@@ -94,4 +118,9 @@ class PostgresFederationContextRepo: 
        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)

        ),

    )

src/federation_manager/adapters/database/partner_lock.py

0 → 100644
+31 −0
Changes for src/federation_manager/adapters/database/partner_lock.py: 31 added lines, 0 removed lines.
Original line number Diff line number Diff line 
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)

src/federation_manager/adapters/database/partner_repo.py

+10 −0
Changes for src/federation_manager/adapters/database/partner_repo.py: 10 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -51,6 +51,15 @@ class PostgresPartnerRepo: 
        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)),
@@ -126,4 +135,5 @@ class PostgresPartnerRepo: 
            base_url=row.base_url,

            name=row.name,

            our_client_secret_set=row.our_client_secret_set,

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

        )