Support Dev Server initiated in-band syncs

inngest/_internal/comm_lib/handler.py

@@ -43,10 +43,6 @@ def __init__(
         framework: server_lib.Framework,
         functions: list[function.Function],
     ) -> None:
-        # TODO: Default to true once in-band syncing is stable
-        self._allow_in_band_sync = env_lib.is_true(
-            const.EnvKey.ALLOW_IN_BAND_SYNC,
-        )
         self._client = client
         self._mode = client._mode
         self._api_origin = client.api_origin
@@ -363,24 +359,21 @@ async def put(
         """Handle a PUT request."""
 
         self._client.logger.info("Syncing app")
+
+        allow_in_band_sync = req.allow_in_band_sync
+        if req.allow_in_band_sync is None:
+            # TODO: Default to true once in-band syncing is stable
+            allow_in_band_sync = env_lib.is_true(
+                const.EnvKey.ALLOW_IN_BAND_SYNC,
+            )
+
         syncer = _Syncer(logger=self._client.logger)
 
         if (
             req.headers.get(server_lib.HeaderKey.SYNC_KIND.value)
             == server_lib.SyncKind.IN_BAND.value
-            and self._allow_in_band_sync
+            and allow_in_band_sync is True
         ):
-            err: typing.Optional[Exception] = None
-            if isinstance(request_signing_key, Exception):
-                err = request_signing_key
-            elif request_signing_key is None:
-                err = Exception("request must be signed for in-band sync")
-            if err is not None:
-                return CommResponse.from_error(
-                    self._client.logger,
-                    err,
-                    status=http.HTTPStatus.UNAUTHORIZED,
-                )
             return syncer.in_band(self, req, request_signing_key)
 
         return await syncer.out_of_band(self, req)
@@ -394,25 +387,21 @@ def put_sync(
         """Handle a PUT request."""
 
         self._client.logger.info("Syncing app")
+
+        allow_in_band_sync = req.allow_in_band_sync
+        if req.allow_in_band_sync is None:
+            # TODO: Default to true once in-band syncing is stable
+            allow_in_band_sync = env_lib.is_true(
+                const.EnvKey.ALLOW_IN_BAND_SYNC,
+            )
+
         syncer = _Syncer(logger=self._client.logger)
 
         if (
             req.headers.get(server_lib.HeaderKey.SYNC_KIND.value)
             == server_lib.SyncKind.IN_BAND.value
-            and self._allow_in_band_sync
+            and allow_in_band_sync is True
         ):
-            err: typing.Optional[Exception] = None
-            if isinstance(request_signing_key, Exception):
-                err = request_signing_key
-            elif request_signing_key is None:
-                err = Exception("request must be signed for in-band sync")
-            if err is not None:
-                return CommResponse.from_error(
-                    self._client.logger,
-                    err,
-                    status=http.HTTPStatus.UNAUTHORIZED,
-                )
-
             return syncer.in_band(self, req, request_signing_key)
 
         return syncer.out_of_band_sync(self, req)
@@ -496,10 +485,19 @@ def in_band(
         req: CommRequest,
         request_signing_key: types.MaybeError[typing.Optional[str]],
     ) -> types.MaybeError[CommResponse]:
-        if not isinstance(request_signing_key, str):
-            # This should be checked earlier, but we'll also check it here since
-            # it's critical
-            return Exception("request must be signed for in-band sync")
+        if handler._signing_key is not None:
+            if isinstance(request_signing_key, Exception):
+                return CommResponse.from_error(
+                    self._logger,
+                    request_signing_key,
+                    status=http.HTTPStatus.UNAUTHORIZED,
+                )
+            if request_signing_key is None:
+                return CommResponse.from_error(
+                    self._logger,
+                    Exception("request must be signed for in-band sync"),
+                    status=http.HTTPStatus.UNAUTHORIZED,
+                )
 
         req_body = server_lib.InBandSynchronizeRequest.from_raw(req.body)
         if isinstance(req_body, Exception):
@@ -522,9 +520,6 @@ def in_band(
         )
         if isinstance(inspection, Exception):
             return inspection
-        if isinstance(inspection, server_lib.UnauthenticatedInspection):
-            # Unreachable
-            return Exception("request must be signed for in-band sync")
 
         res_body = server_lib.InBandSynchronizeResponse(
             app_id=handler._client.app_id,

inngest/_internal/comm_lib/models.py

@@ -14,6 +14,7 @@
 
 
 class CommRequest(types.BaseModel):
+    allow_in_band_sync: typing.Optional[bool]
     body: bytes
     headers: typing.Union[dict[str, str], dict[str, str]]
     query_params: typing.Union[dict[str, str], dict[str, list[str]]]

inngest/_internal/server_lib/registration.py

@@ -8,7 +8,11 @@
 from inngest._internal import const, errors, transforms, types
 
 from .consts import DeployType, Framework
-from .inspection import AuthenticatedInspection, Capabilities
+from .inspection import (
+    AuthenticatedInspection,
+    Capabilities,
+    UnauthenticatedInspection,
+)
 
 
 class _BaseConfig(types.BaseModel):
@@ -185,7 +189,7 @@ class InBandSynchronizeResponse(types.BaseModel):
     env: typing.Optional[str]
     framework: Framework
     functions: list[FunctionConfig]
-    inspection: AuthenticatedInspection
+    inspection: typing.Union[AuthenticatedInspection, UnauthenticatedInspection]
     platform: typing.Optional[str]
     sdk_author: str = const.AUTHOR
     sdk_language: str = const.LANGUAGE

inngest/digital_ocean.py

@@ -17,6 +17,7 @@ def serve(
     client: client_lib.Inngest,
     functions: list[function.Function],
     *,
+    allow_in_band_sync: typing.Optional[bool] = None,
     serve_origin: typing.Optional[str] = None,
     serve_path: typing.Optional[str] = None,
 ) -> typing.Callable[[dict[str, object], _Context], _Response]:
@@ -28,6 +29,7 @@ def serve(
         client: Inngest client.
         functions: List of functions to serve.
 
+        allow_in_band_sync: Whether to allow in-band syncing.
         serve_origin: Origin to serve the functions from.
         serve_path: The entire function path (e.g. /api/v1/web/fn-b094417f/sample/hello).
     """
@@ -71,6 +73,7 @@ def main(event: dict[str, object], context: _Context) -> _Response:
             request_url = urllib.parse.urljoin(context.api_host, path)
 
             comm_req = comm_lib.CommRequest(
+                allow_in_band_sync=allow_in_band_sync,
                 body=_to_body_bytes(http.body),
                 headers=http.headers,
                 query_params=query_params,

inngest/django.py

@@ -28,6 +28,7 @@ def serve(
     client: client_lib.Inngest,
     functions: list[function.Function],
     *,
+    allow_in_band_sync: typing.Optional[bool] = None,
     serve_origin: typing.Optional[str] = None,
     serve_path: typing.Optional[str] = None,
 ) -> django.urls.URLPattern:
@@ -39,7 +40,7 @@ def serve(
         client: Inngest client.
         functions: List of functions to serve.
 
-        async_mode: [DEPRECATED] Whether to serve functions asynchronously.
+        allow_in_band_sync: Whether to allow in-band syncing.
         serve_origin: Origin to serve Inngest from.
         serve_path: Path to serve Inngest from.
     """
@@ -59,13 +60,15 @@ def serve(
         return _create_handler_async(
             client,
             handler,
+            allow_in_band_sync=allow_in_band_sync,
             serve_origin=serve_origin,
             serve_path=serve_path,
         )
     else:
         return _create_handler_sync(
             client,
             handler,
+            allow_in_band_sync=allow_in_band_sync,
             serve_origin=serve_origin,
             serve_path=serve_path,
         )
@@ -75,13 +78,15 @@ def _create_handler_sync(
     client: client_lib.Inngest,
     handler: comm_lib.CommHandler,
     *,
+    allow_in_band_sync: typing.Optional[bool],
     serve_origin: typing.Optional[str],
     serve_path: typing.Optional[str],
 ) -> django.urls.URLPattern:
     def inngest_api(
         request: django.http.HttpRequest,
     ) -> django.http.HttpResponse:
         comm_req = comm_lib.CommRequest(
+            allow_in_band_sync=allow_in_band_sync,
             body=request.body,
             headers=dict(request.headers.items()),
             query_params=dict(request.GET.items()),
@@ -126,6 +131,7 @@ def _create_handler_async(
     client: client_lib.Inngest,
     handler: comm_lib.CommHandler,
     *,
+    allow_in_band_sync: typing.Optional[bool],
     serve_origin: typing.Optional[str],
     serve_path: typing.Optional[str],
 ) -> django.urls.URLPattern:
@@ -143,6 +149,7 @@ async def inngest_api(
         request: django.http.HttpRequest,
     ) -> django.http.HttpResponse:
         comm_req = comm_lib.CommRequest(
+            allow_in_band_sync=allow_in_band_sync,
             body=request.body,
             headers=dict(request.headers.items()),
             query_params=dict(request.GET.items()),

inngest/fast_api.py

@@ -23,6 +23,7 @@ def serve(
     client: client_lib.Inngest,
     functions: list[function.Function],
     *,
+    allow_in_band_sync: typing.Optional[bool] = None,
     serve_origin: typing.Optional[str] = None,
     serve_path: typing.Optional[str] = None,
 ) -> None:
@@ -35,6 +36,7 @@ def serve(
         client: Inngest client.
         functions: List of functions to serve.
 
+        allow_in_band_sync: Whether to allow in-band syncing.
         serve_origin: Origin to serve the functions from.
         serve_path: Path to serve the functions from.
     """
@@ -53,6 +55,7 @@ async def get_api_inngest(
             client,
             handler.get_sync(
                 comm_lib.CommRequest(
+                    allow_in_band_sync=allow_in_band_sync,
                     body=await request.body(),
                     headers=dict(request.headers.items()),
                     query_params=dict(request.query_params.items()),
@@ -72,6 +75,7 @@ async def post_inngest_api(
             client,
             await handler.post(
                 comm_lib.CommRequest(
+                    allow_in_band_sync=allow_in_band_sync,
                     body=await request.body(),
                     headers=dict(request.headers.items()),
                     query_params=dict(request.query_params.items()),
@@ -91,6 +95,7 @@ async def put_inngest_api(
             client,
             await handler.put(
                 comm_lib.CommRequest(
+                    allow_in_band_sync=allow_in_band_sync,
                     body=await request.body(),
                     headers=dict(request.headers.items()),
                     query_params=dict(request.query_params.items()),

inngest/flask.py

@@ -22,6 +22,7 @@ def serve(
     client: client_lib.Inngest,
     functions: list[function.Function],
     *,
+    allow_in_band_sync: typing.Optional[bool] = None,
     serve_origin: typing.Optional[str] = None,
     serve_path: typing.Optional[str] = None,
 ) -> None:
@@ -34,6 +35,7 @@ def serve(
         client: Inngest client.
         functions: List of functions to serve.
 
+        allow_in_band_sync: Whether to allow in-band syncing.
         serve_origin: Origin to serve the functions from.
         serve_path: Path to serve the functions from.
     """
@@ -53,6 +55,7 @@ def serve(
             app,
             client,
             handler,
+            allow_in_band_sync=allow_in_band_sync,
             serve_origin=serve_origin,
             serve_path=serve_path,
         )
@@ -61,6 +64,7 @@ def serve(
             app,
             client,
             handler,
+            allow_in_band_sync=allow_in_band_sync,
             serve_origin=serve_origin,
             serve_path=serve_path,
         )
@@ -71,6 +75,7 @@ def _create_handler_async(
     client: client_lib.Inngest,
     handler: comm_lib.CommHandler,
     *,
+    allow_in_band_sync: typing.Optional[bool],
     serve_origin: typing.Optional[str],
     serve_path: typing.Optional[str],
 ) -> None:
@@ -80,6 +85,7 @@ def _create_handler_async(
     )
     async def inngest_api() -> typing.Union[flask.Response, str]:
         comm_req = comm_lib.CommRequest(
+            allow_in_band_sync=allow_in_band_sync,
             body=_get_body_bytes(),
             headers=dict(flask.request.headers.items()),
             query_params=flask.request.args,
@@ -116,6 +122,7 @@ def _create_handler_sync(
     client: client_lib.Inngest,
     handler: comm_lib.CommHandler,
     *,
+    allow_in_band_sync: typing.Optional[bool],
     serve_origin: typing.Optional[str],
     serve_path: typing.Optional[str],
 ) -> None:
@@ -125,6 +132,7 @@ def _create_handler_sync(
     )
     def inngest_api() -> typing.Union[flask.Response, str]:
         comm_req = comm_lib.CommRequest(
+            allow_in_band_sync=allow_in_band_sync,
             body=_get_body_bytes(),
             headers=dict(flask.request.headers.items()),
             query_params=flask.request.args,

inngest/tornado.py

@@ -23,6 +23,7 @@ def serve(
     client: client_lib.Inngest,
     functions: list[function.Function],
     *,
+    allow_in_band_sync: typing.Optional[bool] = None,
     serve_origin: typing.Optional[str] = None,
     serve_path: typing.Optional[str] = None,
 ) -> None:
@@ -35,6 +36,7 @@ def serve(
         client: Inngest client.
         functions: List of functions to serve.
 
+        allow_in_band_sync: Whether to allow in-band syncing.
         serve_origin: Origin to serve the functions from.
         serve_path: Path to serve the functions from.
     """
@@ -54,6 +56,7 @@ def data_received(
         def get(self) -> None:
             comm_res = handler.get_sync(
                 comm_lib.CommRequest(
+                    allow_in_band_sync=allow_in_band_sync,
                     body=self.request.body,
                     headers=dict(self.request.headers.items()),
                     query_params=_parse_query_params(
@@ -71,6 +74,7 @@ def get(self) -> None:
         def post(self) -> None:
             comm_res = handler.post_sync(
                 comm_lib.CommRequest(
+                    allow_in_band_sync=allow_in_band_sync,
                     body=self.request.body,
                     headers=dict(self.request.headers.items()),
                     query_params=_parse_query_params(
@@ -88,6 +92,7 @@ def post(self) -> None:
         def put(self) -> None:
             comm_res = handler.put_sync(
                 comm_lib.CommRequest(
+                    allow_in_band_sync=allow_in_band_sync,
                     body=self.request.body,
                     headers=dict(self.request.headers.items()),
                     query_params=_parse_query_params(

tests/test_registration/base.py

@@ -28,5 +28,6 @@ def serve(
         self,
         client: inngest.Inngest,
         fns: list[inngest.Function],
+        allow_in_band_sync: typing.Optional[bool] = None,
     ) -> None:
         raise NotImplementedError()