Improve the queue performance by adding an explicit lock mechanism

[jun-he]

Pull Request type

• Bugfix
• Feature
• Refactoring (no functional changes, no api changes)
• Build related changes (Please run ./gradlew build --write-locks to refresh dependencies)
• Other (please describe): performance improvement

NOTE: Please remember to run ./gradlew spotlessApply to fix any format violations.

Changes in this PR

Improve the queue performance by adding an explicit lock mechanism because SKIP LOCKED does not perform well if there are a large number of rows.

[Copilot]

Pull Request Overview

This PR improves queue performance by implementing an explicit database locking mechanism to replace the previous SKIP LOCKED approach when dealing with large numbers of rows. The changes introduce a new locking table structure and modify the dequeue operation to use explicit locks instead of relying solely on database-level row locking.

Key changes:

• Added explicit lock mechanism using a special queue entry (queue_id=0) to control access to queue operations
• Modified dequeue operation to check for locks before processing messages and removed SKIP LOCKED clause
• Enhanced logging in SubworkflowStepRuntime for better debugging and error tracking

Reviewed Changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 5 comments.

File	Description
MaestroQueueDao.java	Implements core locking mechanism with addLock method and updates dequeue logic to use explicit locks
MaestroQueueWorkerService.java	Adds lock initialization during worker service startup
MaestroQueueDaoTest.java	Updates tests to work with new locking mechanism and adds test coverage for addLock functionality
SubworkflowStepRuntime.java	Improves logging with better error messages and workflow identity information

[Copilot]

The nested try-with-resources blocks and conditional logic create complex control flow. Consider extracting the lock checking and message dequeuing logic into separate methods to improve readability and maintainability.

[Copilot]

The magic values '0' for queue_id and owned_until, and 'lock' for payload should be defined as named constants to improve code maintainability and prevent potential errors.

Suggested change

	"SELECT msg_id FROM maestro_queue WHERE queue_id=0 AND owned_until=0 AND msg_id=? FOR UPDATE SKIP LOCKED";
	+ "VALUES (" + LOCK_QUEUE_ID + "," + LOCK_OWNED_UNTIL + ",?,'" + LOCK_PAYLOAD + "',EXTRACT(EPOCH FROM NOW())::INT8*1000) ON CONFLICT DO NOTHING";
	private static final String LOCK_SCAN_QUERY =
	"SELECT msg_id FROM maestro_queue WHERE queue_id=" + LOCK_QUEUE_ID + " AND owned_until=" + LOCK_OWNED_UNTIL + " AND msg_id=? FOR UPDATE SKIP LOCKED";

[Copilot]

The magic value '0' for msgId in the test should be defined as a constant or use a more descriptive approach to create lock removal messages, as this pattern is repeated multiple times in the test.

Suggested change

	dao.remove(new MessageDto(0, String.valueOf(queueId), null, 0));
	dao.remove(new MessageDto(LOCK_REMOVAL_MSG_ID, String.valueOf(queueId), null, 0));
	assertEquals(1, dao.addLock(queueId));
	assertEquals(0, dao.addLock(queueId));
	dao.remove(new MessageDto(LOCK_REMOVAL_MSG_ID, String.valueOf(queueId), null, 0));

[Copilot]

The variable 'idx' is reset inside the while loop but was already declared and used outside the loop. Consider declaring a new variable within the loop scope to avoid confusion and potential bugs.

[Copilot]

The TODO comment indicates incomplete error handling implementation. This should be tracked in an issue or implemented as part of this PR since it affects production code reliability.

Suggested change

	// todo improve this error handling to gracefully clean up all resources
	// Attempt to clean up resources associated with this subworkflow step runtime
	try {
	cleanupResources(workflowSummary, step, runtimeSummary);
	} catch (Exception cleanupEx) {
	LOG.error("Error during resource cleanup for subworkflow step runtime {}{}: {}", workflowSummary.getIdentity(), runtimeSummary.getIdentity(), cleanupEx);
	}