Skip to content

Services Documentation

Detailed documentation for each service component in the Virtufin API.

Table of Contents

GatewayService

File: Services/GatewayService.cs

The central gRPC service providing gateway functionality including service discovery, method invocation, and event streaming.

gRPC Service Definition

service Gateway {
  rpc Invoke (InvokeRequest) returns (InvokeResponse);
  rpc InvokeJson (InvokeJsonRequest) returns (InvokeJsonResponse);
  rpc ListServices (ListServicesRequest) returns (ListServicesResponse);
  rpc ListMethods (ListMethodsRequest) returns (ListMethodsResponse);
  rpc GetMethodSchema (GetMethodSchemaRequest) returns (GetMethodSchemaResponse);
}

Methods

Invoke

Invokes a method on a registered backend service.

public override async Task<InvokeResponse> Invoke(
    InvokeRequest request, 
    ServerCallContext context)

Request: - service - Target service name (e.g., "workmanager") - method - Method name to invoke - requestData - JSON payload for the request

Response: - success - Boolean indicating success - status - gRPC status code string - message - Error message if failed - responseData - JSON response payload

InvokeJson

Invokes a method on a registered backend service using JSON request/response data. The recommended method for most use cases.

public override async Task<InvokeJsonResponse> InvokeJson(
    InvokeJsonRequest request, 
    ServerCallContext context)

Request: - service - Target service name (e.g., "workmanager") - method - Method name to invoke - requestData - JSON payload for the request (as a string)

Response: - success - Boolean indicating success - status - gRPC status code string - message - Error message if failed - responseData - JSON response payload (as a string)

ListServices

Lists all services registered in the gateway configuration.

public override async Task<ListServicesResponse> ListServices(
    ListServicesRequest request, 
    ServerCallContext context)

Response: - services - List of service name strings

ListMethods

Lists all methods available on a specific service.

public override async Task<ListMethodsResponse> ListMethods(
    ListMethodsRequest request, 
    ServerCallContext context)

Request: - service - Service name to query

Response: - methods - List of MethodInfo objects with name, input/output types, streaming flags

Special Handling: For virtufin.Gateway or Gateway, returns the built-in gateway methods directly.

GetMethodSchema

Returns the schema for a specific method including input/output types and field definitions.

public override async Task<GetMethodSchemaResponse> GetMethodSchema(
    GetMethodSchemaRequest request, 
    ServerCallContext context)

Request: - service - Service name - method - Method name - type - Schema type to return (typically "input" or "output")

Response: - schema - MethodSchema object with input_type_name, output_type_name, schema_type, and fields (list of FieldSchema with name, type, type_name, is_repeated, is_optional, oneof_group, oneof_cases)

Used by clients to introspect a service's wire format before calling it.

Note: Gateway has no Subscribe RPC. Event streaming lives on the Pubsub service (Pubsub.Subscribe, topic-filtered only) — see PubsubService and events.md. The WatchSystemEvents RPC was also removed; per-service lifecycle events are delivered through Pubsub.Subscribe on the appropriate per-service topic (e.g. websocketmanager.lifecycle, workmanager.lifecycle).

GrpcReflectionService

File: Services/GrpcReflectionService.cs

Provides dynamic gRPC service discovery and invocation using the gRPC Reflection protocol.

Key Classes

Class Purpose
GrpcReflectionService Public API, manages per-service caches
ServiceDescriptorCache Per-service descriptor caching and loading
ServiceMethodInfo Method metadata storage
MethodSchema Request/response schema
FieldSchema Field definition
GrpcCallResult Invocation result wrapper

Public API

GetMethodsAsync

public async Task<List<ServiceMethodInfo>> GetMethodsAsync(
    string serviceName, 
    CancellationToken cancellationToken = default)

Returns all methods for a service discovered via reflection.

GetMethodSchemaAsync

public async Task<MethodSchema?> GetMethodSchemaAsync(
    string serviceName, 
    string methodName, 
    string? type = null, 
    CancellationToken cancellationToken = default)

Returns the schema for a method's input or output type.

Parameters: - serviceName - Service to query - methodName - Method name - type - "input" or "output" (default: input)

ExecuteCallAsync

public async Task<GrpcCallResult> ExecuteCallAsync(
    string serviceName,
    string methodName,
    string requestJson,
    Dictionary<string, string>? metadata = null,
    CancellationToken cancellationToken = default)

Executes a gRPC call by name using reflection data.

Returns: GrpcCallResult with: - Success - Boolean - StatusCode - gRPC status - Message - Error detail - Data - Response JSON

GetFileDescriptorProtosAsync

public async Task<List<ByteString>> GetFileDescriptorProtosAsync(
    string serviceName, 
    CancellationToken cancellationToken = default)

Returns raw protocol buffer descriptors for the service.

Internal Processing

ServiceDescriptorCache

Manages the cached state for a single service:

  • Lazy Loading: Descriptors loaded on first access
  • Thread-Safe: Uses locking for initialization
  • Descriptor Building: Parses FileDescriptorProtos and builds indexes
  • Cache staleness: The cache has no TTL and no invalidation mechanism. If a backend service redeploys with a changed proto schema, the gateway will serve stale descriptors until it restarts. This is a known limitation (tracked as P2-1). In production, restart the gateway after any backend service redeploy that changes proto schemas. A configurable TTL (REFLECTION_CACHE_TTL_SECONDS) and an invalidation endpoint are planned.

JSON↔Protobuf Conversion

The service handles conversion between JSON and protobuf:

Request (JSON → Protobuf): 1. Try JsonParser.Default.Parse() (preferred) 2. Fall back to field-by-field parsing

Response (Protobuf → JSON): 1. Try msgDesc.Parser.ParseFrom() 2. Fall back to field-by-field binary reading

Well-Known Descriptors

Includes descriptors for: - google.protobuf.FileDescriptor (for proto schema) - google.api.Http (for REST annotations) - google.api.Annotations (for HTTP rules)

These are needed because reflection servers don't return transitive dependencies.

GrpcChannelPool

File: Services/GrpcChannelPool.cs

Thread-safe pool of gRPC channels organized by service name.

Public API

public GrpcChannel GetChannel(string serviceName)

Gets or creates a gRPC channel for the specified service.

Parameters: - serviceName - Service name from configuration

Returns: GrpcChannel for the service

Throws: ArgumentException if service not found in configuration

Channel Management

  • Lazy Creation: Channels created on first access
  • Reuse: Same channel returned for repeated calls to same service
  • Lifecycle: All channels disposed on Dispose()

Configuration Integration

Channels are created using service configuration:

services:
  - name: workmanager
    protocol: grpc
    grpc:
      host: localhost
      port: 5002
    pubsub:
      pubsubName: pubsub
    state:
      storeName: statestore
    jobs:
      crons: []

Hard requirement: Every service in the configuration must implement gRPC reflection. When a gRPC channel is created for a service, the API Gateway probes for reflection support. If reflection is not available, an InvalidOperationException is thrown — the channel is not created and the service cannot be proxied. Ensure all backend services register gRPC reflection via Grpc.AspNetCore.Server.Reflection.

Channel address: http://{host}:{port}

PubsubService

File: Services/PubSubService.cs

Provides pub/sub messaging operations using Dapr pub/sub. Implements the Pubsub gRPC service defined in Protos/pubsub.proto.

gRPC Service Definition

service Pubsub {
  rpc PublishEvent (PublishRequest) returns (PublishResponse);
  rpc Subscribe (PubsubSubscribeRequest) returns (stream PubsubSubscribeResponse);
  rpc Unsubscribe (UnsubscribeRequest) returns (UnsubscribeResponse);
}

Methods

PublishEvent

Publishes an event to a Dapr topic.

public override async Task<PublishResponse> PublishEvent(
    PublishRequest request,
    ServerCallContext context)

Request: - topic - The topic name to publish to - cloudevent - The CloudEvent message (CloudNative.CloudEvents.V1.CloudEvent proto)

Response: - success - Boolean indicating success - status - Status string ("OK" or "ERROR") - message - Success/error message

Subscribe

Subscribes to a topic and streams events as they arrive. Topics are created dynamically (no configuration needed).

public override async Task Subscribe(
    PubsubSubscribeRequest request,
    IServerStreamWriter<PubsubSubscribeResponse> responseStream,
    ServerCallContext context)

Request: - topic - The topic name to subscribe to

Response: Server-streaming of PubsubSubscribeResponse messages containing: - topic - The topic name - cloudevent - The CloudEvent message (CloudNative.CloudEvents.V1.CloudEvent proto) - message_id - Unique message identifier - timestamp - ISO 8601 timestamp

Unsubscribe

Stops receiving events for a subscription.

public override async Task<UnsubscribeResponse> Unsubscribe(
    UnsubscribeRequest request,
    ServerCallContext context)

Request: - subscription_id - The subscription ID returned from Subscribe

Response: - success - Boolean indicating success - status - Status string - message - Status message

PubsubSubscriptionManager

File: Services/PubSubService.cs (inner class PubsubSubscriptionManager)

Manages pub/sub subscriptions and broadcasts messages to subscribers. Inherits from SubscriptionManagerBase.

Public API

// Subscribe to a topic
string Subscribe(string topic, IServerStreamWriter<PubsubSubscribeResponse> writer, CancellationToken cancellationToken)

// Unsubscribe by ID
void UnsubscribeById(string subscriptionId)

// Broadcast to topic
Task BroadcastToTopicAsync(string topic, byte[] data, string? messageId = null)

Key Features

  • Inherits from SubscriptionManagerBase<PubsubSubscribeResponse, PubsubSubscription>
  • Uses scope system: scopes are pubsub/{topic}
  • 5-minute idle timeout per subscription
  • Automatic cleanup on cancellation

ConfigService

File: Services/ConfigService.cs

Exposes the service registry configuration — which services are registered, their gRPC endpoints, and their pubsub / state component names — to admin and observability clients. Implements the Config gRPC service defined in Protos/config.proto.

One caveat on what this reports. dapr_app_id was removed from ConfigService: routing is a direct gRPC dial to grpc.host/grpc.port, never Dapr service invocation. pubsub.pubsub_name is honoured, but as a gateway-wide setting rather than a per-service one — see events.md.

gRPC Service Definition

service Config {
  rpc ListServices (ListConfigServicesRequest) returns (ListConfigServicesResponse);
  rpc GetService (GetConfigServiceRequest) returns (GetConfigServiceResponse);
}

Methods

ListServices

Returns all services registered in services.json, each with its Dapr app-id, gRPC host/port, pubsub component name, state store name, and scheduled cron jobs.

Response: ListConfigServicesResponse.services — list of ConfigService protos with name, grpc.{host,port}, pubsub.pubsub_name, state.store_name, jobs.crons[].

GetService

Returns a single service by name, or an empty Service field if not found.

Request: GetConfigServiceRequest.name — service name to look up.

Response: GetConfigServiceResponse.service — ConfigService proto, or unset if the service is not registered.

StateService

File: Services/StateService.cs

Provides state management using the Dapr state store. Implements the State gRPC service defined in Protos/state.proto.

gRPC Service Definition

service State {
  rpc SaveState (SaveStateRequest) returns (SaveStateResponse);
  rpc GetState (GetStateRequest) returns (GetStateResponse);
  rpc DeleteState (DeleteStateRequest) returns (DeleteStateResponse);
  rpc QueryState (QueryStateRequest) returns (QueryStateResponse);
  rpc GetBulkState (GetBulkStateRequest) returns (GetBulkStateResponse);
  rpc DeleteBulkState (DeleteBulkStateRequest) returns (DeleteBulkStateResponse);
}

message SaveStateRequest {
  string service = 1;
  string key = 2;
  string value = 3;        // MUST be valid JSON
  string etag = 4;
  bool publish_change = 5; // publish a state.change event after the write
  bool include_value = 6;  // include the value in that event
  int32 ttl_seconds = 7;
  bool create_only = 8;    // succeed only if the key does not already exist
}

See proto/state.proto for the full message set.

What service means

service selects the Dapr state store to use, via a lookup in the gateway's services.json. It must be non-empty and must name a registered service; the match is exact and case-sensitive, so WorkManager is rejected where workmanager is accepted.

service is not an isolation boundary. The API does not check that key begins with service, QueryState filters are not scoped to the caller's prefix, and all registered services currently share one store. Keys follow the {service}.{entity}.{id} convention by caller-side discipline — see AGENTS.md for the full convention and its limits.

REST Endpoints

  • POST /v1/state/save-state — save state
  • GET /v1/state/{service}/{key} — get state for a key
  • DELETE /v1/state/{service}/{key} — delete state
  • GET /v1/state/{service}/query — query state
  • GET /v1/state/{service}/bulk — get many keys
  • DELETE /v1/state/{service}/bulk — delete many keys

ETag and Optimistic Concurrency

Scenario Behavior
No ETag provided Last-write-wins. The write always succeeds.
Matching ETag Write succeeds. State is updated.
Mismatched ETag Fails with StatusCode.Aborted. Re-read and retry.
create_only: true, empty ETag Succeeds only if the key does not exist; otherwise Aborted.

An ETag of "0" is rejected with InvalidArgument — it is the Redis component's force-write sentinel and would silently defeat the compare-and-swap.

var current = await client.GetStateAsync("workmanager", "workmanager.worker.abc123");
await client.SaveStateAsync("workmanager", "workmanager.worker.abc123", newValue,
                            etag: current.Etag);
// Throws RpcException with StatusCode.Aborted if the ETag doesn't match.

QueryState

QueryState takes Dapr's query-state JSON DSL and passes it to the store:

{"filter": {"PREFIX": "workmanager.worker."}}
{"filter": {"EQ": {"key": "workmanager.worker.abc123"}}}

It requires a store whose Dapr component implements the Query API. For Redis/Valkey that means the RediSearch module must be loaded and the component must declare queryIndexes metadata; without both, every query fails — including Dapr's "{}" match-everything query. Use GetBulkState against known keys as the fallback.

limit is applied by the gateway after the store returns its full result set, so a broad filter against a large store still materializes every match in gateway memory.

State Change Broadcasting

SaveState, DeleteState and DeleteBulkState publish a change event only when the request sets publish_change: true. Events go to the single global topic state.change, not to a per-service topic; the payload carries service, key (or keys), and action ("save" or "delete").

Setting include_value: true adds the value to that payload — and therefore broadcasts it to every subscriber of state.change, platform-wide. See events.md.

Error reporting

StateService splits its error channel, which is worth knowing when writing a client:

RPC Failure surfaces as
GetState, GetBulkState, QueryState thrown RpcException
SaveState, DeleteState, DeleteBulkState gRPC status OK, status.success == false

Check status.success on writes; a client that only catches RpcException will read a failed save as a success.

TriggerService

File: Services/TriggerService.cs

Wraps Dapr's Scheduler API to schedule recurring cron triggers. When a trigger fires, DaprAppCallbackService publishes an empty CloudEvent to the trigger's target_topic via Dapr pub/sub.

gRPC Service Definition

service Trigger {
  rpc ScheduleTrigger (ScheduleTriggerRequest) returns (ScheduleTriggerResponse);
  rpc GetTrigger (GetTriggerRequest) returns (GetTriggerResponse);
  rpc DeleteTrigger (DeleteTriggerRequest) returns (DeleteTriggerResponse);
  rpc ListTriggers (ListTriggersRequest) returns (ListTriggersResponse);
}

Key Behavior

  • schedule is a 6-field cron expression or an @every duration.
  • Errors are returned in-band as StatusResponse { success: false } with a gRPC status of OK — check status.success, not the gRPC status code.
  • Trigger delivery arrives on Dapr's AppCallback / OnJobEvent route, implemented by DaprAppCallbackService and bound to the stable route by DaprStableAppCallbackMethodProvider.