estuary-cdk: implement webhook server logic

Implements a `webhook` submodule that includes

- A `start_webhook_server` function
- A `WebhookDocument` model that automatically enriches document
  metadata
- A `match` submodule with message matching logic

Signed-off-by: nicolaslazo <45973144+nicolaslazo@users.noreply.github.com>

Author: nicolaslazo

estuary-cdk/common.Dockerfile

@@ -51,4 +51,6 @@ ENV DOCS_URL=${DOCS_URL}
 ENV ENCRYPTION_URL=${ENCRYPTION_URL}
 ENV CONNECTOR_NAME=${CONNECTOR_NAME}
 
+EXPOSE 8080
+
 CMD ["/bin/sh", "-c", "/opt/venv/bin/python -m $(echo \"$CONNECTOR_NAME\" | tr '-' '_')"]

estuary-cdk/estuary_cdk/capture/base_capture_connector.py

@@ -3,8 +3,9 @@
 import json
 import os
 import sys
+from collections.abc import Awaitable
 from datetime import UTC, datetime, timedelta
-from typing import Any, Awaitable, BinaryIO, Callable, Generic
+from typing import Any, BinaryIO, Callable, Generic
 
 from pydantic import BaseModel
 
@@ -36,6 +37,7 @@
 PERIODIC_RESTART_INTERVAL = 24 * 60 * 60  # 24 hours
 GRACEFUL_SHUTDOWN_TIMEOUT = 30 * 60  # 30 minutes
 TASK_CANCELLATION_TIMEOUT = 5 * 60  # 5 minutes
+WEBHOOK_SHUTDOWN_TIMEOUT = 60  # 1 minute
 
 
 class TerminateTaskGroup(Exception):
@@ -52,6 +54,7 @@ class BaseCaptureConnector(
     BaseConnector[Request[EndpointConfig, ResourceConfig, _ConnectorState]],
     HTTPMixin,
     Generic[EndpointConfig, ResourceConfig, _ConnectorState],
+    metaclass=abc.ABCMeta,
 ):
     output: BinaryIO = sys.stdout.buffer
 
@@ -138,7 +141,7 @@ async def handle(
             opened, capture = await self.open(log, open)
             await self._emit(Response(opened=opened))
 
-            stopping = Task.Stopping(asyncio.Event())
+            stopping = Task.Stopping()
 
             async def periodic_stop() -> None:
                 """Trigger graceful shutdown after 24 hours of continuous operation."""
@@ -152,7 +155,7 @@ async def enforce_shutdown(tg: asyncio.TaskGroup) -> None:
                 This handles shutdown triggered by any source: periodic_stop,
                 stream exceptions, or any other code that sets stopping.event.
                 """
-                await stopping.event.wait()
+                _ = await stopping.event.wait()
 
                 log.debug("Waiting for graceful exit.")
 
@@ -236,6 +239,20 @@ async def enforce_shutdown(tg: asyncio.TaskGroup) -> None:
             except* TerminateTaskGroup:
                 pass  # Expected when enforce_shutdown terminates the task group
 
+            # TaskGroup exited = all non-webhook tasks done.
+            # Signal webhook server to drain and shut down.
+            if stopping.webhook_task:
+                stopping.webhook_event.set()
+                try:
+                    await asyncio.wait_for(
+                        stopping.webhook_task, timeout=WEBHOOK_SHUTDOWN_TIMEOUT
+                    )
+                except asyncio.TimeoutError:
+                    log.error(
+                        f"Webhook server did not shut down in {WEBHOOK_SHUTDOWN_TIMEOUT // 60} minutes. Forcing exit."
+                    )
+                    os._exit(1)
+
             # Cancel the background tasks if capture exited gracefully
             # before the shutdown timeout.
             _ = periodic_stop_task.cancel()

estuary-cdk/estuary_cdk/capture/common.py

@@ -1,43 +1,42 @@
 import abc
 import asyncio
 import functools
-from enum import Enum, StrEnum
+import inspect
+from collections.abc import AsyncGenerator, Awaitable
 from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
-import inspect
+from enum import Enum, StrEnum
+from itertools import combinations
 from logging import Logger
 from typing import (
     Any,
-    AsyncGenerator,
-    Awaitable,
     Callable,
-    cast,
     ClassVar,
     Generic,
-    Iterable,
     Literal,
     TypeVar,
+    cast,
 )
 
 from pydantic import AwareDatetime, BaseModel, Field, NonNegativeInt
 
 from ..cron import next_fire
-from ..utils import json_merge_patch
 from ..flow import (
     AccessToken,
+    AuthorizationCodeFlowOAuth2Credentials,
     BaseOAuth2Credentials,
+    BasicAuth,
     CaptureBinding,
     ClientCredentialsOAuth2Credentials,
-    OAuth2TokenFlowSpec,
-    AuthorizationCodeFlowOAuth2Credentials,
     LongLivedClientCredentialsOAuth2Credentials,
+    OAuth2Spec,
+    OAuth2TokenFlowSpec,
     ResourceOwnerPasswordOAuth2Credentials,
     RotatingOAuth2Credentials,
-    OAuth2Spec,
     ValidationError,
-    BasicAuth,
 )
 from ..pydantic_polyfill import GenericModel
+from ..utils import json_merge_patch
 from . import Task, request, response
 
 LogCursor = tuple[str | int] | AwareDatetime | NonNegativeInt
@@ -113,10 +112,10 @@ def pop_cursor_marker(cursor: dict) -> dict:
 and "no pages remain" in a response context.
 """
 
-
 class Triggers(Enum):
     BACKFILL = "BACKFILL"
 
+
 @dataclass
 class SourcedSchema:
     """
@@ -279,7 +278,8 @@ class Snapshot(BaseModel, extra="forbid"):
     )
 
     is_connector_initiated: bool = Field(
-        default=False, description="Indicates if this backfill was initiated by the connector.",
+        default=False,
+        description="Indicates if this backfill was initiated by the connector.",
     )
 
 
@@ -530,11 +530,30 @@ def resolve_bindings(
     resources: list[Resource[Any, _BaseResourceConfig, Any]],
     resource_term="Resource",
 ) -> list[tuple[_ResolvableBinding, Resource[Any, _BaseResourceConfig, Any]]]:
+
     resolved: list[
         tuple[_ResolvableBinding, Resource[Any, _BaseResourceConfig, Any]]
     ] = []
     errors: list[str] = []
 
+    # Check for routing conflicts between webhook match rules.
+    from .webhook.resources import WebhookResourceConfig
+
+    webhook_match_rules = [
+        resource.initial_config.match_rule
+        for resource in resources
+        if isinstance(resource.initial_config, WebhookResourceConfig)
+    ]
+    errors.extend(
+        filter(
+            bool,
+            (
+                rule.list_compatibility_errors(other)
+                for rule, other in combinations(webhook_match_rules, 2)
+            ),
+        )
+    )
+
     for binding in bindings:
         path = binding.resourceConfig.path()
 
@@ -698,11 +717,18 @@ async def _run(task: Task):
             if inspect.iscoroutine(result):
                 await result
 
-
         if soonest_future_scheduled_initialization:
             # Gracefully exit to ensure relatively close adherence to any bindings'
             # re-initialization schedules.
-            asyncio.create_task(scheduled_stop(task, soonest_future_scheduled_initialization))
+            asyncio.create_task(
+                scheduled_stop(task, soonest_future_scheduled_initialization)
+            )
+
+        if task.stopping.webhook_task is not None:
+            # The webhook server runs outside the TaskGroup so it can outlive
+            # non-webhook tasks during two-phase shutdown. Block here so the
+            # TaskGroup stays alive until graceful shutdown begins.
+            await task.stopping.event.wait()
 
     return (response.Opened(explicitAcknowledgements=False), _run)
 

estuary-cdk/estuary_cdk/capture/task.py

@@ -1,18 +1,19 @@
-from dataclasses import dataclass
-import decimal
-from estuary_cdk.capture.connector_status import ConnectorStatus
-from pydantic import Field
-from typing import Generic, Awaitable, Any, BinaryIO, Callable
-import orjson
-import base64
-from logging import Logger
 import asyncio
+import base64
+import decimal
 import tempfile
 import traceback
+from collections.abc import Awaitable
+from dataclasses import dataclass, field
+from logging import Logger
+from typing import Any, BinaryIO, Callable, Generic
+
+import orjson
 import xxhash
+from pydantic import Field
+
+from estuary_cdk.capture.connector_status import ConnectorStatus
 
-from . import request, response
-from ._emit import emit_from_buffer
 from ..flow import (
     ConnectorSpec,
     ConnectorState,
@@ -21,6 +22,8 @@
     ResourceConfig,
 )
 from ..pydantic_polyfill import GenericModel
+from . import request, response
+from ._emit import emit_from_buffer
 
 
 class Request(GenericModel, Generic[EndpointConfig, ResourceConfig, ConnectorState]):
@@ -85,9 +88,15 @@ class Stopping:
         The Task's coroutine should monitor this event and exit when it's set AND
         it has no more immediate work to do (for example, no further documents are
         currently ready to be captured).
+
+        `webhook_event` is a separate event for the webhook server. It is set
+        after all non-webhook tasks have completed, signaling the webhook server
+        to reject new requests with 503 and clean up.
         """
 
-        event: asyncio.Event
+        event: asyncio.Event = field(default_factory=asyncio.Event)
+        webhook_event: asyncio.Event = field(default_factory=asyncio.Event)
+        webhook_task: asyncio.Task[None] | None = None
         first_error: Exception | None = None
         first_error_task: str | None = None