[STREAM-640] Push-Based Broker / Worker POC Protos

.gitignore

@@ -149,7 +149,7 @@ data/
 .artifacts/
 
 # Generated bindings
-/py/sentry_protos/*
+# /py/sentry_protos/*
 
 # Generated documentation
 /docs/build/*

proto/sentry_protos/taskbroker/v1/taskbroker.proto

@@ -82,8 +82,26 @@ service ConsumerService {
 
   // Update the state of a task with execution results.
   rpc SetTaskStatus(SetTaskStatusRequest) returns (SetTaskStatusResponse) {}
+
+  // Add a worker to the broker's inner worker pool.
+  rpc AddWorker(AddWorkerRequest) returns (AddWorkerResponse) {}
+
+  // Remove a worker from the broker's inner worker pool.
+  rpc RemoveWorker(RemoveWorkerRequest) returns (RemoveWorkerResponse) {}
 }
 
+message AddWorkerRequest {
+  string address = 1;
+}
+
+message AddWorkerResponse {}
+
+message RemoveWorkerRequest {
+  string address = 1;
+}
+
+message RemoveWorkerResponse {}
+
 message GetTaskRequest {
   optional string namespace = 1;
 }
@@ -104,6 +122,8 @@ message SetTaskStatusRequest {
 
   // If fetch_next is provided, receive a new task in the response
   optional FetchNextTask fetch_next_task = 3;
+
+  string address = 4;
 }
 
 message SetTaskStatusResponse {

proto/sentry_protos/taskworker/v1/taskworker.proto

@@ -0,0 +1,20 @@
+syntax = "proto3";
+
+package sentry_protos.taskworker.v1;
+
+import "sentry_protos/taskbroker/v1/taskbroker.proto";
+
+service WorkerService {
+  // Receives a task to execute.
+  rpc PushTask(PushTaskRequest) returns (PushTaskResponse) {}
+}
+
+message PushTaskRequest {
+  sentry_protos.taskbroker.v1.TaskActivation task = 1;
+  string callback_url = 2;
+}
+
+message PushTaskResponse {
+  bool added = 1;
+  uint32 queue_size = 2;
+}

py/sentry_protos/__init__.py

@@ -0,0 +1 @@
+__version__ = "0.4.10"

py/sentry_protos/conduit/v1alpha/publish_pb2.pyi

@@ -0,0 +1,90 @@
+"""
+@generated by mypy-protobuf.  Do not edit manually!
+isort:skip_file
+"""
+
+import builtins
+import google.protobuf.descriptor
+import google.protobuf.internal.enum_type_wrapper
+import google.protobuf.message
+import google.protobuf.struct_pb2
+import google.protobuf.timestamp_pb2
+import sys
+import typing
+
+if sys.version_info >= (3, 10):
+    import typing as typing_extensions
+else:
+    import typing_extensions
+
+DESCRIPTOR: google.protobuf.descriptor.FileDescriptor
+
+class _Phase:
+    ValueType = typing.NewType("ValueType", builtins.int)
+    V: typing_extensions.TypeAlias = ValueType
+
+class _PhaseEnumTypeWrapper(google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[_Phase.ValueType], builtins.type):
+    DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor
+    PHASE_UNSPECIFIED: _Phase.ValueType  # 0
+    PHASE_START: _Phase.ValueType  # 1
+    """Initial event to establish stream"""
+    PHASE_DELTA: _Phase.ValueType  # 2
+    """Incremental updates"""
+    PHASE_END: _Phase.ValueType  # 3
+    """Clean stream termination"""
+    PHASE_ERROR: _Phase.ValueType  # 4
+    """Stream encountered an error"""
+
+class Phase(_Phase, metaclass=_PhaseEnumTypeWrapper): ...
+
+PHASE_UNSPECIFIED: Phase.ValueType  # 0
+PHASE_START: Phase.ValueType  # 1
+"""Initial event to establish stream"""
+PHASE_DELTA: Phase.ValueType  # 2
+"""Incremental updates"""
+PHASE_END: Phase.ValueType  # 3
+"""Clean stream termination"""
+PHASE_ERROR: Phase.ValueType  # 4
+"""Stream encountered an error"""
+global___Phase = Phase
+
+@typing.final
+class PublishRequest(google.protobuf.message.Message):
+    DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+    CHANNEL_ID_FIELD_NUMBER: builtins.int
+    MESSAGE_ID_FIELD_NUMBER: builtins.int
+    CLIENT_TIMESTAMP_FIELD_NUMBER: builtins.int
+    PHASE_FIELD_NUMBER: builtins.int
+    SEQUENCE_FIELD_NUMBER: builtins.int
+    PAYLOAD_FIELD_NUMBER: builtins.int
+    channel_id: builtins.str
+    """Channel identifier for routing"""
+    message_id: builtins.str
+    """Unique message identifier for deduplication and idempotency"""
+    phase: global___Phase.ValueType
+    """Current phase of the streaming lifecycle"""
+    sequence: builtins.int
+    """Monotonically increasing sequence number within a stream"""
+    @property
+    def client_timestamp(self) -> google.protobuf.timestamp_pb2.Timestamp:
+        """Timestamp when client sent the event"""
+
+    @property
+    def payload(self) -> google.protobuf.struct_pb2.Struct:
+        """Application-specific payload data"""
+
+    def __init__(
+        self,
+        *,
+        channel_id: builtins.str = ...,
+        message_id: builtins.str = ...,
+        client_timestamp: google.protobuf.timestamp_pb2.Timestamp | None = ...,
+        phase: global___Phase.ValueType = ...,
+        sequence: builtins.int = ...,
+        payload: google.protobuf.struct_pb2.Struct | None = ...,
+    ) -> None: ...
+    def HasField(self, field_name: typing.Literal["client_timestamp", b"client_timestamp", "payload", b"payload"]) -> builtins.bool: ...
+    def ClearField(self, field_name: typing.Literal["channel_id", b"channel_id", "client_timestamp", b"client_timestamp", "message_id", b"message_id", "payload", b"payload", "phase", b"phase", "sequence", b"sequence"]) -> None: ...
+
+global___PublishRequest = PublishRequest

py/sentry_protos/conduit/v1alpha/publish_pb2_grpc.py

@@ -0,0 +1,29 @@
+# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
+"""Client and server classes corresponding to protobuf-defined services."""
+import grpc
+import warnings
+
+
+GRPC_GENERATED_VERSION = '1.65.4'
+GRPC_VERSION = grpc.__version__
+EXPECTED_ERROR_RELEASE = '1.66.0'
+SCHEDULED_RELEASE_DATE = 'August 6, 2024'
+_version_not_supported = False
+
+try:
+    from grpc._utilities import first_version_is_lower
+    _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION)
+except ImportError:
+    _version_not_supported = True
+
+if _version_not_supported:
+    warnings.warn(
+        f'The grpc package installed is at version {GRPC_VERSION},'
+        + f' but the generated code in sentry_protos/conduit/v1alpha/publish_pb2_grpc.py depends on'
+        + f' grpcio>={GRPC_GENERATED_VERSION}.'
+        + f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}'
+        + f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.'
+        + f' This warning will become an error in {EXPECTED_ERROR_RELEASE},'
+        + f' scheduled for release on {SCHEDULED_RELEASE_DATE}.',
+        RuntimeWarning
+    )

py/sentry_protos/conduit/v1alpha/publish_pb2_grpc.pyi

@@ -0,0 +1,17 @@
+"""
+@generated by mypy-protobuf.  Do not edit manually!
+isort:skip_file
+"""
+
+import abc
+import collections.abc
+import grpc
+import grpc.aio
+import typing
+
+_T = typing.TypeVar("_T")
+
+class _MaybeAsyncIterator(collections.abc.AsyncIterator[_T], collections.abc.Iterator[_T], metaclass=abc.ABCMeta): ...
+
+class _ServicerContext(grpc.ServicerContext, grpc.aio.ServicerContext):  # type: ignore[misc, type-arg]
+    ...