fix(abdm): add concurrency lock, optimize hiu-poll query, and add composite indexes
This commit is contained in:
parent
fcce8970fd
commit
2f6b346e25
2
.gitignore
vendored
2
.gitignore
vendored
|
|
@ -70,6 +70,8 @@ writable/uploads/*
|
|||
writable/debugbar/*
|
||||
!writable/debugbar/index.html
|
||||
|
||||
writable/locks/*
|
||||
|
||||
writable/ayushman/*
|
||||
|
||||
php_errors.log
|
||||
|
|
|
|||
|
|
@ -19,19 +19,39 @@ class AbdmHiuPoll extends BaseCommand
|
|||
|
||||
public function run(array $params)
|
||||
{
|
||||
$limit = (int) (CLI::getOption('limit') ?? 30);
|
||||
if ($limit <= 0) {
|
||||
$limit = 30;
|
||||
$lockDir = WRITEPATH . 'locks';
|
||||
if (! is_dir($lockDir)) {
|
||||
@mkdir($lockDir, 0755, true);
|
||||
}
|
||||
$lockFile = $lockDir . DIRECTORY_SEPARATOR . 'abdm_hiu_poll.lock';
|
||||
$lockFp = @fopen($lockFile, 'c+');
|
||||
|
||||
if (! $lockFp || ! flock($lockFp, LOCK_EX | LOCK_NB)) {
|
||||
CLI::write('Another instance of abdm:hiu-poll is already running. Skipping execution.', 'yellow');
|
||||
if ($lockFp) {
|
||||
fclose($lockFp);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
$service = new M3HiuWorkflowService();
|
||||
$summary = $service->pollNatGateway($limit);
|
||||
try {
|
||||
$limit = (int) (CLI::getOption('limit') ?? 30);
|
||||
if ($limit <= 0) {
|
||||
$limit = 30;
|
||||
}
|
||||
|
||||
CLI::write('ABDM HIU NAT Poll Summary', 'yellow');
|
||||
CLI::write('Processed: ' . (int) ($summary['processed'] ?? 0));
|
||||
CLI::write('Consent updates: ' . (int) ($summary['consent_updates'] ?? 0), 'green');
|
||||
CLI::write('Data updates: ' . (int) ($summary['data_updates'] ?? 0), 'green');
|
||||
CLI::write('Failed: ' . (int) ($summary['failed'] ?? 0), ((int) ($summary['failed'] ?? 0) > 0 ? 'red' : 'green'));
|
||||
CLI::write('Skipped: ' . (int) ($summary['skipped'] ?? 0));
|
||||
$service = new M3HiuWorkflowService();
|
||||
$summary = $service->pollNatGateway($limit);
|
||||
|
||||
CLI::write('ABDM HIU NAT Poll Summary', 'yellow');
|
||||
CLI::write('Processed: ' . (int) ($summary['processed'] ?? 0));
|
||||
CLI::write('Consent updates: ' . (int) ($summary['consent_updates'] ?? 0), 'green');
|
||||
CLI::write('Data updates: ' . (int) ($summary['data_updates'] ?? 0), 'green');
|
||||
CLI::write('Failed: ' . (int) ($summary['failed'] ?? 0), ((int) ($summary['failed'] ?? 0) > 0 ? 'red' : 'green'));
|
||||
CLI::write('Skipped: ' . (int) ($summary['skipped'] ?? 0));
|
||||
} finally {
|
||||
flock($lockFp, LOCK_UN);
|
||||
fclose($lockFp);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,18 +19,38 @@ class AbdmHiuRetry extends BaseCommand
|
|||
|
||||
public function run(array $params)
|
||||
{
|
||||
$limit = (int) (CLI::getOption('limit') ?? 20);
|
||||
if ($limit <= 0) {
|
||||
$limit = 20;
|
||||
$lockDir = WRITEPATH . 'locks';
|
||||
if (! is_dir($lockDir)) {
|
||||
@mkdir($lockDir, 0755, true);
|
||||
}
|
||||
$lockFile = $lockDir . DIRECTORY_SEPARATOR . 'abdm_hiu_retry.lock';
|
||||
$lockFp = @fopen($lockFile, 'c+');
|
||||
|
||||
if (! $lockFp || ! flock($lockFp, LOCK_EX | LOCK_NB)) {
|
||||
CLI::write('Another instance of abdm:hiu-retry is already running. Skipping execution.', 'yellow');
|
||||
if ($lockFp) {
|
||||
fclose($lockFp);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
$service = new M3HiuWorkflowService();
|
||||
$summary = $service->retryDue($limit);
|
||||
try {
|
||||
$limit = (int) (CLI::getOption('limit') ?? 20);
|
||||
if ($limit <= 0) {
|
||||
$limit = 20;
|
||||
}
|
||||
|
||||
CLI::write('ABDM M3 HIU Retry Summary', 'yellow');
|
||||
CLI::write('Processed: ' . (int) ($summary['processed'] ?? 0));
|
||||
CLI::write('Success: ' . (int) ($summary['success'] ?? 0), 'green');
|
||||
CLI::write('Failed: ' . (int) ($summary['failed'] ?? 0), ((int) ($summary['failed'] ?? 0) > 0 ? 'red' : 'green'));
|
||||
CLI::write('Skipped: ' . (int) ($summary['skipped'] ?? 0));
|
||||
$service = new M3HiuWorkflowService();
|
||||
$summary = $service->retryDue($limit);
|
||||
|
||||
CLI::write('ABDM M3 HIU Retry Summary', 'yellow');
|
||||
CLI::write('Processed: ' . (int) ($summary['processed'] ?? 0));
|
||||
CLI::write('Success: ' . (int) ($summary['success'] ?? 0), 'green');
|
||||
CLI::write('Failed: ' . (int) ($summary['failed'] ?? 0), ((int) ($summary['failed'] ?? 0) > 0 ? 'red' : 'green'));
|
||||
CLI::write('Skipped: ' . (int) ($summary['skipped'] ?? 0));
|
||||
} finally {
|
||||
flock($lockFp, LOCK_UN);
|
||||
fclose($lockFp);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,50 @@
|
|||
<?php
|
||||
|
||||
namespace App\Database\Migrations;
|
||||
|
||||
use CodeIgniter\Database\Migration;
|
||||
|
||||
class AddCompositeIndexesToAbdmHiuWorkflows extends Migration
|
||||
{
|
||||
public function up(): void
|
||||
{
|
||||
if (! $this->db->tableExists('abdm_hiu_workflows')) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (! $this->indexExists('abdm_hiu_workflows', 'idx_abdm_hiu_op_status_id')) {
|
||||
$this->db->query('CREATE INDEX idx_abdm_hiu_op_status_id ON abdm_hiu_workflows (operation, status, id)');
|
||||
}
|
||||
|
||||
if (! $this->indexExists('abdm_hiu_workflows', 'idx_abdm_hiu_op_status_created')) {
|
||||
$this->db->query('CREATE INDEX idx_abdm_hiu_op_status_created ON abdm_hiu_workflows (operation, status, created_at)');
|
||||
}
|
||||
}
|
||||
|
||||
public function down(): void
|
||||
{
|
||||
if (! $this->db->tableExists('abdm_hiu_workflows')) {
|
||||
return;
|
||||
}
|
||||
|
||||
if ($this->indexExists('abdm_hiu_workflows', 'idx_abdm_hiu_op_status_id')) {
|
||||
$this->db->query('DROP INDEX idx_abdm_hiu_op_status_id ON abdm_hiu_workflows');
|
||||
}
|
||||
|
||||
if ($this->indexExists('abdm_hiu_workflows', 'idx_abdm_hiu_op_status_created')) {
|
||||
$this->db->query('DROP INDEX idx_abdm_hiu_op_status_created ON abdm_hiu_workflows');
|
||||
}
|
||||
}
|
||||
|
||||
private function indexExists(string $table, string $indexName): bool
|
||||
{
|
||||
$rows = $this->db->query('SHOW INDEX FROM ' . $table)->getResultArray();
|
||||
foreach ($rows as $row) {
|
||||
if (($row['Key_name'] ?? '') === $indexName) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
|
@ -199,11 +199,36 @@ class M3HiuWorkflowService
|
|||
}
|
||||
|
||||
$summary = ['processed' => 0, 'consent_updates' => 0, 'data_updates' => 0, 'failed' => 0, 'skipped' => 0];
|
||||
$consentRows = $this->db->table('abdm_hiu_workflows')
|
||||
->select('*')
|
||||
|
||||
$fields = $this->db->getFieldNames('abdm_hiu_workflows') ?? [];
|
||||
$selectCols = ['id', 'request_id', 'transaction_id', 'consent_id', 'abha_address', 'hfr_id', 'workflow_state', 'status', 'created_at', 'updated_at'];
|
||||
foreach (['abdm_consent_request_id', 'abdm_consent_artifact_id', 'completed_at', 'expired_at', 'revoked_at'] as $col) {
|
||||
if (in_array($col, $fields, true)) {
|
||||
$selectCols[] = $col;
|
||||
}
|
||||
}
|
||||
|
||||
$builder = $this->db->table('abdm_hiu_workflows')
|
||||
->select(implode(', ', $selectCols))
|
||||
->where('operation', 'consent_request')
|
||||
->where('status', 'success')
|
||||
->orderBy('id', 'DESC')
|
||||
->whereNotIn('workflow_state', ['COMPLETED', 'EXPIRED', 'REVOKED', 'DENIED']);
|
||||
|
||||
if (in_array('completed_at', $fields, true)) {
|
||||
$builder->where('completed_at IS NULL');
|
||||
}
|
||||
if (in_array('expired_at', $fields, true)) {
|
||||
$builder->where('expired_at IS NULL');
|
||||
}
|
||||
if (in_array('revoked_at', $fields, true)) {
|
||||
$builder->where('revoked_at IS NULL');
|
||||
}
|
||||
|
||||
// Only poll consent requests created within the last 7 days to avoid infinite polling of stale/ancient rows
|
||||
$cutoffDate = date('Y-m-d H:i:s', strtotime('-7 days'));
|
||||
$builder->where('created_at >=', $cutoffDate);
|
||||
|
||||
$consentRows = $builder->orderBy('id', 'DESC')
|
||||
->get($limit)
|
||||
->getResultArray();
|
||||
|
||||
|
|
@ -215,6 +240,25 @@ class M3HiuWorkflowService
|
|||
continue;
|
||||
}
|
||||
$seenConsentRequests[$requestId] = true;
|
||||
|
||||
$consentRowId = (int) ($row['id'] ?? 0);
|
||||
|
||||
// Fast-check: if a child reconcile/data_fetch workflow for this request_id has already reached terminal state
|
||||
$terminalCheck = $this->db->table('abdm_hiu_workflows')
|
||||
->select('id, workflow_state, operation')
|
||||
->where('request_id', $requestId)
|
||||
->whereIn('workflow_state', ['COMPLETED', 'EXPIRED', 'REVOKED', 'DENIED'])
|
||||
->orderBy('id', 'DESC')
|
||||
->get(1)
|
||||
->getRowArray();
|
||||
|
||||
if (! empty($terminalCheck)) {
|
||||
$termState = (string) ($terminalCheck['workflow_state'] ?? '');
|
||||
$this->markConsentRequestTerminal($consentRowId, $termState);
|
||||
$summary['skipped']++;
|
||||
continue;
|
||||
}
|
||||
|
||||
$summary['processed']++;
|
||||
|
||||
$payload = [
|
||||
|
|
@ -237,6 +281,19 @@ class M3HiuWorkflowService
|
|||
if (! $cascade['granted']) {
|
||||
$summary['skipped']++;
|
||||
}
|
||||
|
||||
$consentState = strtolower(trim((string) (
|
||||
$consentResult['workflow_state']
|
||||
?? $consentResult['status']
|
||||
?? $consentResult['consent_status']
|
||||
?? ''
|
||||
)));
|
||||
|
||||
if (in_array($consentState, ['revoked', 'denied', 'expired'], true)) {
|
||||
$this->markConsentRequestTerminal($consentRowId, strtoupper($consentState));
|
||||
} elseif ($cascade['granted'] && $cascade['failed'] === 0 && $cascade['data_updates'] > 0) {
|
||||
$this->markConsentRequestTerminal($consentRowId, 'COMPLETED');
|
||||
}
|
||||
} else {
|
||||
$summary['failed']++;
|
||||
}
|
||||
|
|
@ -245,6 +302,32 @@ class M3HiuWorkflowService
|
|||
return $summary;
|
||||
}
|
||||
|
||||
private function markConsentRequestTerminal(int $consentRowId, string $terminalState): void
|
||||
{
|
||||
if ($consentRowId <= 0 || ! $this->db->tableExists('abdm_hiu_workflows')) {
|
||||
return;
|
||||
}
|
||||
|
||||
$now = Time::now('Asia/Kolkata')->toDateTimeString();
|
||||
$update = [
|
||||
'workflow_state' => $terminalState,
|
||||
'updated_at' => $now,
|
||||
];
|
||||
|
||||
$fields = $this->db->getFieldNames('abdm_hiu_workflows') ?? [];
|
||||
if (in_array('completed_at', $fields, true) && $terminalState === 'COMPLETED') {
|
||||
$update['completed_at'] = $now;
|
||||
}
|
||||
if (in_array('expired_at', $fields, true) && $terminalState === 'EXPIRED') {
|
||||
$update['expired_at'] = $now;
|
||||
}
|
||||
if (in_array('revoked_at', $fields, true) && in_array($terminalState, ['REVOKED', 'DENIED'], true)) {
|
||||
$update['revoked_at'] = $now;
|
||||
}
|
||||
|
||||
$this->db->table('abdm_hiu_workflows')->where('id', $consentRowId)->update($update);
|
||||
}
|
||||
|
||||
/**
|
||||
* Shared GRANTED-consent cascade: once a consent_reconcile call confirms
|
||||
* GRANTED, this fetches decrypted data for EVERY known consent artifact
|
||||
|
|
|
|||
|
|
@ -36,7 +36,7 @@ $BLOCK_START
|
|||
* * * * * cd $PROJECT_DIR && $PHP_BIN spark abdm:push-sync --limit=20 --worker=cron-abdm-m2 >> $LOG_DIR/abdm_push_sync.log 2>&1
|
||||
* * * * * cd $PROJECT_DIR && $PHP_BIN spark bridge:sync --limit 10 >> $LOG_DIR/bridge_sync.log 2>&1
|
||||
* * * * * cd $PROJECT_DIR && $PHP_BIN spark snomed:process-coding >> $LOG_DIR/snomed_process.log 2>&1
|
||||
*/2 * * * * cd $PROJECT_DIR && $PHP_BIN spark abdm:hiu-poll --limit=30 >> $LOG_DIR/abdm_hiu_poll.log 2>&1
|
||||
*/2 * * * * cd $PROJECT_DIR && $PHP_BIN spark abdm:hiu-poll --limit=15 >> $LOG_DIR/abdm_hiu_poll.log 2>&1
|
||||
*/5 * * * * cd $PROJECT_DIR && $PHP_BIN spark abdm:hiu-retry --limit=20 >> $LOG_DIR/abdm_hiu_retry.log 2>&1
|
||||
30 2 * * * cd $PROJECT_DIR && $PHP_BIN spark abdm:prune-logs --days=30 --strip-payload-days=2 --optimize >> $LOG_DIR/abdm_prune_logs.log 2>&1
|
||||
$BLOCK_END
|
||||
|
|
|
|||
|
|
@ -110,7 +110,7 @@ $BLOCK_START
|
|||
# SNOMED CT Medical Coding Worker (Every 1 minute)
|
||||
* * * * * cd $project_dir && $php_bin spark snomed:process-coding >> $log_dir/snomed_process.log 2>&1
|
||||
# ABDM M3 HIU Data Request Poll (Every 2 minutes)
|
||||
*/2 * * * * cd $project_dir && $php_bin spark abdm:hiu-poll --limit=30 >> $log_dir/abdm_hiu_poll.log 2>&1
|
||||
*/2 * * * * cd $project_dir && $php_bin spark abdm:hiu-poll --limit=15 >> $log_dir/abdm_hiu_poll.log 2>&1
|
||||
# ABDM M3 HIU Retry Queue (Every 5 minutes)
|
||||
*/5 * * * * cd $project_dir && $php_bin spark abdm:hiu-retry --limit=20 >> $log_dir/abdm_hiu_retry.log 2>&1
|
||||
# ABDM Audit & Gateway Log Maintenance (Daily at 02:30 AM)
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user