Discovery manifest v3 (#103)

Author: slinkydeveloper

python/restate/discovery.py

@@ -60,21 +60,36 @@ def __init__(self, setContentTypeIfEmpty: bool, contentType: Optional[str] = Non
         self.jsonSchema = jsonSchema
 
 class Handler:
-    def __init__(self, name: str, ty: Optional[ServiceHandlerType] = None, input: Optional[InputPayload | Dict[str, str]] = None, output: Optional[OutputPayload] = None, description: Optional[str] = None, metadata: Optional[Dict[str, str]] = None):
+    # pylint: disable=R0902
+    def __init__(self, name: str, ty: Optional[ServiceHandlerType] = None, input: Optional[InputPayload | Dict[str, str]] = None, output: Optional[OutputPayload] = None, description: Optional[str] = None, metadata: Optional[Dict[str, str]] = None, inactivityTimeout: Optional[int] = None, abortTimeout: Optional[int] = None, journalRetention: Optional[int] = None, idempotencyRetention: Optional[int] = None, workflowCompletionRetention: Optional[int] = None, enableLazyState: Optional[bool] = None, ingressPrivate: Optional[bool] = None):
         self.name = name
         self.ty = ty
         self.input = input
         self.output = output
         self.documentation = description
         self.metadata = metadata
+        self.inactivityTimeout = inactivityTimeout
+        self.abortTimeout = abortTimeout
+        self.journalRetention = journalRetention
+        self.idempotencyRetention = idempotencyRetention
+        self.workflowCompletionRetention = workflowCompletionRetention
+        self.enableLazyState = enableLazyState
+        self.ingressPrivate = ingressPrivate
 
 class Service:
-    def __init__(self, name: str, ty: ServiceType, handlers: List[Handler], description: Optional[str] = None, metadata: Optional[Dict[str, str]] = None):
+    # pylint: disable=R0902
+    def __init__(self, name: str, ty: ServiceType, handlers: List[Handler], description: Optional[str] = None, metadata: Optional[Dict[str, str]] = None, inactivityTimeout: Optional[int] = None, abortTimeout: Optional[int] = None, journalRetention: Optional[int] = None, idempotencyRetention: Optional[int] = None, enableLazyState: Optional[bool] = None, ingressPrivate: Optional[bool] = None):
         self.name = name
         self.ty = ty
         self.handlers = handlers
         self.documentation = description
         self.metadata = metadata
+        self.inactivityTimeout = inactivityTimeout
+        self.abortTimeout = abortTimeout
+        self.journalRetention = journalRetention
+        self.idempotencyRetention = idempotencyRetention
+        self.enableLazyState = enableLazyState
+        self.ingressPrivate = ingressPrivate
 
 class Endpoint:
     def __init__(self, protocolMode: ProtocolMode, minProtocolVersion: int, maxProtocolVersion: int, services: List[Service]):
@@ -148,20 +163,51 @@ def json_schema_from_type_hint(type_hint: Optional[TypeHint[Any]]) -> Any:
     return type_hint_to_json_schema(type_hint.annotation)
 
 
-
+# pylint: disable=R0912
 def compute_discovery_json(endpoint: RestateEndpoint,
                            version: int,
-                           discovered_as: typing.Literal["bidi", "request_response"]) -> typing.Tuple[typing.Dict[str, str] ,str]:
+                           discovered_as: typing.Literal["bidi", "request_response"]) -> str:
     """
-    return restate's discovery object as JSON 
+    return restate's discovery object as JSON
     """
-    if version != 1:
-        raise ValueError(f"Unsupported protocol version {version}")
 
     ep = compute_discovery(endpoint, discovered_as)
+
+    # Validate that new discovery fields aren't used with older protocol versions
+    if version <= 2:
+        # Check for new discovery fields in version 3 that shouldn't be used in version 2 or lower
+        for service in ep.services:
+            if service.inactivityTimeout is not None:
+                raise ValueError("inactivityTimeout is only supported in discovery protocol version 3")
+            if service.abortTimeout is not None:
+                raise ValueError("abortTimeout is only supported in discovery protocol version 3")
+            if service.idempotencyRetention is not None:
+                raise ValueError("idempotencyRetention is only supported in discovery protocol version 3")
+            if service.journalRetention is not None:
+                raise ValueError("journalRetention is only supported in discovery protocol version 3")
+            if service.enableLazyState is not None:
+                raise ValueError("enableLazyState is only supported in discovery protocol version 3")
+            if service.ingressPrivate is not None:
+                raise ValueError("ingressPrivate is only supported in discovery protocol version 3")
+
+            for handler in service.handlers:
+                if handler.inactivityTimeout is not None:
+                    raise ValueError("inactivityTimeout is only supported in discovery protocol version 3")
+                if handler.abortTimeout is not None:
+                    raise ValueError("abortTimeout is only supported in discovery protocol version 3")
+                if handler.idempotencyRetention is not None:
+                    raise ValueError("idempotencyRetention is only supported in discovery protocol version 3")
+                if handler.journalRetention is not None:
+                    raise ValueError("journalRetention is only supported in discovery protocol version 3")
+                if handler.workflowCompletionRetention is not None:
+                    raise ValueError("workflowCompletionRetention is only supported in discovery protocol version 3")
+                if handler.enableLazyState is not None:
+                    raise ValueError("enableLazyState is only supported in discovery protocol version 3")
+                if handler.ingressPrivate is not None:
+                    raise ValueError("ingressPrivate is only supported in discovery protocol version 3")
+
     json_str = json.dumps(ep, cls=PythonClassEncoder, allow_nan=False)
-    headers = {"content-type": "application/vnd.restate.endpointmanifest.v1+json"}
-    return (headers, json_str)
+    return json_str
 
 
 def compute_discovery(endpoint: RestateEndpoint, discovered_as : typing.Literal["bidi", "request_response"]) -> Endpoint:
@@ -200,11 +246,28 @@ def compute_discovery(endpoint: RestateEndpoint, discovered_as : typing.Literal[
                                             input=inp,
                                             output=out,
                                             description=handler.description,
-                                            metadata=handler.metadata))
+                                            metadata=handler.metadata,
+                                            inactivityTimeout=int(handler.inactivity_timeout.total_seconds() * 1000) if handler.inactivity_timeout else None,
+                                            abortTimeout=int(handler.abort_timeout.total_seconds() * 1000) if handler.abort_timeout else None,
+                                            journalRetention=int(handler.journal_retention.total_seconds() * 1000) if handler.journal_retention else None,
+                                            idempotencyRetention=int(handler.idempotency_retention.total_seconds() * 1000) if handler.idempotency_retention else None,
+                                            workflowCompletionRetention=int(handler.workflow_retention.total_seconds() * 1000) if handler.workflow_retention else None,
+                                            enableLazyState=handler.enable_lazy_state,
+                                            ingressPrivate=handler.ingress_private))
         # add the service
         description = service.service_tag.description
         metadata = service.service_tag.metadata
-        services.append(Service(name=service.name, ty=service_type, handlers=service_handlers, description=description, metadata=metadata))
+        services.append(Service(name=service.name,
+                               ty=service_type,
+                               handlers=service_handlers,
+                               description=description,
+                               metadata=metadata,
+                               inactivityTimeout=int(service.inactivity_timeout.total_seconds() * 1000) if service.inactivity_timeout else None,
+                               abortTimeout=int(service.abort_timeout.total_seconds() * 1000) if service.abort_timeout else None,
+                               journalRetention=int(service.journal_retention.total_seconds() * 1000) if service.journal_retention else None,
+                               idempotencyRetention=int(service.idempotency_retention.total_seconds() * 1000) if service.idempotency_retention else None,
+                               enableLazyState=service.enable_lazy_state if hasattr(service, 'enable_lazy_state') else None,
+                               ingressPrivate=service.ingress_private))
 
     if endpoint.protocol:
         protocol_mode = PROTOCOL_MODES[endpoint.protocol]

python/restate/handler.py

@@ -16,6 +16,7 @@
 """
 
 from dataclasses import dataclass
+from datetime import timedelta
 from inspect import Signature
 from typing import Any, Callable, Awaitable, Dict, Generic, Literal, Optional, TypeVar
 
@@ -104,6 +105,23 @@ def update_handler_io_with_type_hints(handler_io: HandlerIO[I, O], signature: Si
 class Handler(Generic[I, O]):
     """
     Represents a handler for a service.
+
+    Attributes:
+        service_tag: The service tag for the handler.
+        handler_io: The input/output configuration for the handler.
+        kind: The kind of handler (exclusive, shared, workflow).
+        name: The name of the handler.
+        fn: The handler function.
+        arity: The number of parameters in the handler function.
+        description: Documentation for this handler definition.
+        metadata: Custom metadata for this handler definition.
+        inactivity_timeout: Inactivity timeout duration.
+        abort_timeout: Abort timeout duration.
+        journal_retention: Journal retention duration.
+        idempotency_retention: Idempotency retention duration.
+        workflow_retention: Workflow completion retention duration.
+        enable_lazy_state: If true, lazy state is enabled.
+        ingress_private: If true, the handler cannot be invoked from the HTTP nor Kafka ingress.
     """
     service_tag: ServiceTag
     handler_io: HandlerIO[I, O]
@@ -113,21 +131,52 @@ class Handler(Generic[I, O]):
     arity: int
     description: Optional[str] = None
     metadata: Optional[Dict[str, str]] = None
+    inactivity_timeout: Optional[timedelta] = None
+    abort_timeout: Optional[timedelta] = None
+    journal_retention: Optional[timedelta] = None
+    idempotency_retention: Optional[timedelta] = None
+    workflow_retention: Optional[timedelta] = None
+    enable_lazy_state: Optional[bool] = None
+    ingress_private: Optional[bool] = None
 
 
 # disable too many arguments warning
 # pylint: disable=R0913
-
+# pylint: disable=R0914
 def make_handler(service_tag: ServiceTag,
                  handler_io: HandlerIO[I, O],
                  name: str | None,
                  kind: Optional[Literal["exclusive", "shared", "workflow"]],
                  wrapped: Any,
                  signature: Signature,
                  description: Optional[str] = None,
-                 metadata: Optional[Dict[str, str]] = None) -> Handler[I, O]:
+                 metadata: Optional[Dict[str, str]] = None,
+                 inactivity_timeout: Optional[timedelta] = None,
+                 abort_timeout: Optional[timedelta] = None,
+                 journal_retention: Optional[timedelta] = None,
+                 idempotency_retention: Optional[timedelta] = None,
+                 workflow_retention: Optional[timedelta] = None,
+                 enable_lazy_state: Optional[bool] = None,
+                 ingress_private: Optional[bool] = None) -> Handler[I, O]:
     """
     Factory function to create a handler.
+
+    Args:
+        service_tag: The service tag for the handler.
+        handler_io: The input/output configuration for the handler.
+        name: The name of the handler.
+        kind: The kind of handler (exclusive, shared, workflow).
+        wrapped: The wrapped function.
+        signature: The signature of the function.
+        description: Documentation for this handler definition.
+        metadata: Custom metadata for this handler definition.
+        inactivity_timeout: Inactivity timeout duration.
+        abort_timeout: Abort timeout duration.
+        journal_retention: Journal retention duration.
+        idempotency_retention: Idempotency retention duration.
+        workflow_retention: Workflow completion retention duration.
+        enable_lazy_state: If true, lazy state is enabled.
+        ingress_private: If true, the handler cannot be invoked from the HTTP nor Kafka ingress.
     """
     # try to deduce the handler name
     handler_name = name
@@ -149,7 +198,14 @@ def make_handler(service_tag: ServiceTag,
                             fn=wrapped,
                             arity=arity,
                             description=description,
-                            metadata=metadata)
+                            metadata=metadata,
+                            inactivity_timeout=inactivity_timeout,
+                            abort_timeout=abort_timeout,
+                            journal_retention=journal_retention,
+                            idempotency_retention=idempotency_retention,
+                            workflow_retention=workflow_retention,
+                            enable_lazy_state=enable_lazy_state,
+                            ingress_private=ingress_private)
 
     vars(wrapped)[RESTATE_UNIQUE_HANDLER_SYMBOL] = handler
     return handler