Data Acquisition Worker Service

Overview

The worker builds on the Data Acquisition service, which connects and queries a tenant's endpoint for FHIR resources that are needed to evaluate patients for a measure.

• When a log entry is marked ready, DataAcquisitionLogService.StartRetrievalProcess updates its status and produces a ReadyToAcquire event for the log and facility.
• Program.cs registers hosted services such as ReadyToAcquireListener, RetryListener, and RetryScheduleService to consume these events and provide retry behavior if needed.
• Inside ReadyToAcquireListener, each message triggers PatientDataService.ExecuteLogRequest, which retrieves the patient data associated with the log and facility.
• As the service acquires FHIR resources, it will store them in a cache destination for downstream services to use for processing (Normalization, Evaluation). Depending on the configuration, the service is capable of caching to Azure Blob Storage and Redis.
• After all patient resources have been acquired and cached, the service produces a ResourcesAcquired event that contains metadata involving the patient, facility, reporting period and information on the destination of the cached resources.

flowchart LR
nDataAcquisitionWorkerService_1C09C8B6["Service: Data Acquisition Worker Service"]
  nDataAcquisitionWorkerService_1C09C8B6 -->|produces| nPatientListsAcquired_4681892E["Event: PatientListsAcquired"]
  nDataAcquisitionWorkerService_1C09C8B6 -->|produces| nCernerPatientsAcquired_3467767F["Event: CernerPatientsAcquired"]
  nDataAcquisitionWorkerService_1C09C8B6 -->|produces| nresources_acquired_566EB692["Event: resources-acquired"]
  nReadyToAcquire_7C2F519C["Event: ReadyToAcquire"] -->|consumed by| nDataAcquisitionWorkerService_1C09C8B6
  subgraph nDataAccess_2A523756_domain["Domain: Data Access"]
    nCernerPatientsAcquired_3467767F
    nDataAcquisitionWorkerService_1C09C8B6
    nPatientListsAcquired_4681892E
    nReadyToAcquire_7C2F519C
    nresources_acquired_566EB692
  end

Common Configurations

• Azure App Configuration
• Kafka Configuration
• Kafka Consumer Retry Configuration
• Service Registry Configuration
• CORS Configuration
• Token Service Configuration
• Service Authentication
• SQL Server Database Configuration

Service Configurations

• ResourceCache__CacheImplementation: Acquired FHIR resources a stored in a cache destination. The Data Acquisition Worker will store to different desintations depending on how this is set. The following cache implementation settings are supported:
  • ABS: All acquired FHIR resources will be cached in Azure Blob Storage.
  • Redis: All acquired FHIR resources will be cached in Redis.
  • Hybrid: Acquired FHIR resources will be primarily cached in Redis. If capacity limits are met to store into Redis, the Data Acquisition service will store the resources into Azure Blob Storage.
• ResourceCache__Redis__ConnectionString: Connection string to access the deployed Redis environment.
• ResourceCache__Redis__Password: Password for the deployed Redis environment.
• ResourceCache__Redis__MemoryThresholdPercent: When CacheImplementation is set to Hybrid, the worker service will primarily store resources into Redis until it exceeds this threshold setting. When the capacity exceeds the set setting, the worker service will store resources into Azure Blob Storage. The defailt configuration of this setting is 80.0.
• ResourceCache__BlobStorage__ConnectionString: Connection string to access the deployed Azure Blob Storage environment.
• ResourceCache__BlobStorage__BlobContainerName: The Azure Blob Storage container name where cached resources will be stored.
• ResourceCache__BlobStorage__BlobRoot: Root Azure Blob Storage directory the cached resources will be located.

Features and Functionality

The Data Acquisition Worker is responsible for the actual execution of FHIR queries defined by the Data Acquisition Service.

Log Execution Flow

1. Event Consumption: The worker listens for ReadyToAcquire events. Each event contains a LogId and FacilityId.
2. Log Claiming: When an event is received, the worker attempts to atomically transition the log's status from Ready to Queued. This ensures that even if multiple worker instances receive the same event, only one will process it.
3. Background Processing: Once claimed, the work item is added to an internal Channel, which is processed by the AcquisitionProcessorBackgroundService.
4. Resource Retrieval:
   • The worker retrieves the full DataAcquisitionLog and its associated QueryPlan details.
   • It executes the FHIR search queries against the tenant's endpoint.
   • It handles pagination for large result sets.
   • For Reference queries, it scans previously acquired resources for the specified references and fetches them.
5. Resource Caching: As resources are acquired, they are cached in either Redis or Azure Blob Storage. Where they are stored depends on the app setting of the worker service.
6. Completion: Once all resources for the log are processed, the log status is updated to Completed, and a ResourcesAcquired event is published.

Concurrency and Performance

The worker is designed for high-throughput data acquisition:

• Bounded Channel: The internal work queue has a configurable capacity (default 200) to provide backpressure.
• Parallel Execution: The AcquisitionProcessorBackgroundService uses Parallel.ForEachAsync with a configurable MaxDegreeOfParallelism (default 8) to process multiple acquisition logs concurrently.
• Throttling: The worker respects EHR-specific throttling limits if configured in the tenant's Query Config.

Error Handling and Retries

• Transient Failures: If a query fails due to a network error or EHR timeout, the worker catches the exception and updates the log status to Pending. The DataAcquisitionService will then reschedule it for retry.
• Poison Messages: If a message cannot be processed after multiple attempts, it is moved to a Dead Letter Queue (DLQ).
• Logging: All significant actions and errors are recorded in the Notes field of the DataAcquisitionLog in the database, providing a detailed audit trail for developers and support staff.

OpenAPI Operations

Relationships

flowchart LR
nDataAcquisitionWorkerService_1C09C8B6["Service: Data Acquisition Worker Service"]
  nDataAcquisitionWorkerService_1C09C8B6 -->|produces| nPatientListsAcquired_4681892E["Event: PatientListsAcquired"]
  nDataAcquisitionWorkerService_1C09C8B6 -->|produces| nCernerPatientsAcquired_3467767F["Event: CernerPatientsAcquired"]
  nDataAcquisitionWorkerService_1C09C8B6 -->|produces| nresources_acquired_566EB692["Event: resources-acquired"]
  nReadyToAcquire_7C2F519C["Event: ReadyToAcquire"] -->|consumed by| nDataAcquisitionWorkerService_1C09C8B6
  subgraph nDataAccess_2A523756_domain["Domain: Data Access"]
    nCernerPatientsAcquired_3467767F
    nDataAcquisitionWorkerService_1C09C8B6
    nPatientListsAcquired_4681892E
    nReadyToAcquire_7C2F519C
    nresources_acquired_566EB692
  end