Skip to content

Fix non-deterministic ordering in message polling queries #15

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Copilot wants to merge 4 commits into from
Closed

Fix non-deterministic ordering in message polling queries #15

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0fd6159
Initial plan
Copilot Feb 7, 2026
fcf7e7b
Merge origin/main to see H2 2.4.240 changes
Copilot Feb 7, 2026
bcb307c
Fix non-deterministic ordering in all database adapters
Copilot Feb 7, 2026
92122b0
Update indexes to include id column for optimal query performance
Copilot Feb 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions delayedqueue-jvm/src/main/kotlin/org/funfix/delayedqueue/jvm/internals/jdbc/h2/H2Adapter.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,7 @@ internal class H2Adapter(driver: JdbcDriver, tableName: String) :
"createdAt"
FROM "$tableName"
WHERE "pKind" = ? AND "scheduledAt" <= ?
ORDER BY "scheduledAt"
ORDER BY "scheduledAt", "id"
LIMIT 1
FOR UPDATE
"""
Expand Down Expand Up @@ -136,7 +136,7 @@ internal class H2Adapter(driver: JdbcDriver, tableName: String) :
SELECT TOP $limit "id"
FROM "$tableName"
WHERE "pKind" = ? AND "scheduledAt" <= ?
ORDER BY "scheduledAt"
ORDER BY "scheduledAt", "id"
)
"""

Expand Down
37 changes: 35 additions & 2 deletions ...edqueue-jvm/src/main/kotlin/org/funfix/delayedqueue/jvm/internals/jdbc/h2/H2Migrations.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,11 +31,44 @@ internal object H2Migrations {
ON "$tableName" ("pKind", "pKey");

CREATE INDEX "${tableName}__KindPlusScheduledAtIndex"
ON "$tableName" ("pKind", "scheduledAt");
ON "$tableName" ("pKind", "scheduledAt", "id");

CREATE INDEX "${tableName}__LockUuidPlusIdIndex"
ON "$tableName" ("lockUuid", "id");
""",
)
),
// Migration 2: Update index to include 'id' column for deterministic ordering
Migration(
sql =
"""
DROP INDEX IF EXISTS "${tableName}__KindPlusScheduledAtIndex";
Copy link
Copy Markdown
Member

@alexandru alexandru Feb 7, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@copilot This isn't deployed in any production systems yet, I don't want a second migration.

CREATE INDEX "${tableName}__KindPlusScheduledAtIndex"
ON "$tableName" ("pKind", "scheduledAt", "id");
""",
needsExecution = { conn ->
// Check if index exists but doesn't have the 'id' column
val metadata = conn.underlying.metaData
var hasIndex = false
var hasIdColumn = false

metadata.getIndexInfo(null, null, tableName, false, false).use { rs ->
while (rs.next()) {
val indexName = rs.getString("INDEX_NAME")
if (indexName == "${tableName}__KindPlusScheduledAtIndex") {
hasIndex = true
val columnName = rs.getString("COLUMN_NAME")
val ordinalPosition = rs.getShort("ORDINAL_POSITION")
// Check if 'id' is in position 3
if (columnName == "id" && ordinalPosition == 3.toShort()) {
hasIdColumn = true
}
}
}
}

// Run migration if index exists but doesn't have id column
hasIndex && !hasIdColumn
},
),
)
}
4 changes: 2 additions & 2 deletions ...ue-jvm/src/main/kotlin/org/funfix/delayedqueue/jvm/internals/jdbc/hsqldb/HSQLDBAdapter.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ internal class HSQLDBAdapter(driver: JdbcDriver, tableName: String) :
"createdAt"
FROM "$tableName"
WHERE "pKind" = ? AND "scheduledAt" <= ?
ORDER BY "scheduledAt"
ORDER BY "scheduledAt", "id"
FETCH FIRST 1 ROWS ONLY
"""

Expand Down Expand Up @@ -112,7 +112,7 @@ internal class HSQLDBAdapter(driver: JdbcDriver, tableName: String) :
SELECT "id"
FROM "$tableName"
WHERE "pKind" = ? AND "scheduledAt" <= ?
ORDER BY "scheduledAt"
ORDER BY "scheduledAt", "id"
LIMIT $limit
)
"""
Expand Down
37 changes: 35 additions & 2 deletions ...jvm/src/main/kotlin/org/funfix/delayedqueue/jvm/internals/jdbc/hsqldb/HSQLDBMigrations.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,11 +31,44 @@ internal object HSQLDBMigrations {
ON "$tableName" ("pKind", "pKey");

CREATE INDEX "${tableName}__KindPlusScheduledAtIndex"
ON "$tableName" ("pKind", "scheduledAt");
ON "$tableName" ("pKind", "scheduledAt", "id");

CREATE INDEX "${tableName}__LockUuidPlusIdIndex"
ON "$tableName" ("lockUuid", "id");
""",
)
),
// Migration 2: Update index to include 'id' column for deterministic ordering
Migration(
sql =
"""
DROP INDEX "${tableName}__KindPlusScheduledAtIndex" IF EXISTS;
CREATE INDEX "${tableName}__KindPlusScheduledAtIndex"
ON "$tableName" ("pKind", "scheduledAt", "id");
""",
needsExecution = { conn ->
// Check if index exists but doesn't have the 'id' column
val metadata = conn.underlying.metaData
var hasIndex = false
var hasIdColumn = false

metadata.getIndexInfo(null, null, tableName, false, false).use { rs ->
while (rs.next()) {
val indexName = rs.getString("INDEX_NAME")
if (indexName == "${tableName}__KindPlusScheduledAtIndex") {
hasIndex = true
val columnName = rs.getString("COLUMN_NAME")
val ordinalPosition = rs.getShort("ORDINAL_POSITION")
// Check if 'id' is in position 3
if (columnName == "id" && ordinalPosition == 3.toShort()) {
hasIdColumn = true
}
}
}
}

// Run migration if index exists but doesn't have id column
hasIndex && !hasIdColumn
},
),
)
}
4 changes: 2 additions & 2 deletions ...-jvm/src/main/kotlin/org/funfix/delayedqueue/jvm/internals/jdbc/mariadb/MariaDBAdapter.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,7 @@ internal class MariaDBAdapter(driver: JdbcDriver, tableName: String) :
`createdAt`
FROM `$tableName`
WHERE `pKind` = ? AND `scheduledAt` <= ?
ORDER BY `scheduledAt`
ORDER BY `scheduledAt`, `id`
LIMIT 1
FOR UPDATE SKIP LOCKED
"""
Expand Down Expand Up @@ -133,7 +133,7 @@ internal class MariaDBAdapter(driver: JdbcDriver, tableName: String) :
SELECT `id`
FROM `$tableName`
WHERE `pKind` = ? AND `scheduledAt` <= ?
ORDER BY `scheduledAt`
ORDER BY `scheduledAt`, `id`
LIMIT $limit
FOR UPDATE SKIP LOCKED
) AS subq
Expand Down
36 changes: 34 additions & 2 deletions ...m/src/main/kotlin/org/funfix/delayedqueue/jvm/internals/jdbc/mariadb/MariaDBMigrations.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,9 +29,41 @@ internal object MariaDBMigrations {
UNIQUE KEY `${tableName}__KindPlusKeyUniqueIndex` (`pKind`, `pKey`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE INDEX `${tableName}__KindPlusScheduledAtIndex` ON `$tableName`(`pKind`, `scheduledAt`);
CREATE INDEX `${tableName}__KindPlusScheduledAtIndex` ON `$tableName`(`pKind`, `scheduledAt`, `id`);
CREATE INDEX `${tableName}__LockUuidPlusIdIndex` ON `$tableName`(`lockUuid`, `id`)
""",
)
),
// Migration 2: Update index to include 'id' column for deterministic ordering
Migration(
sql =
"""
DROP INDEX IF EXISTS `${tableName}__KindPlusScheduledAtIndex` ON `$tableName`;
CREATE INDEX `${tableName}__KindPlusScheduledAtIndex` ON `$tableName`(`pKind`, `scheduledAt`, `id`);
""",
needsExecution = { conn ->
// Check if index exists but doesn't have the 'id' column
val metadata = conn.underlying.metaData
var hasIndex = false
var hasIdColumn = false

metadata.getIndexInfo(null, null, tableName, false, false).use { rs ->
while (rs.next()) {
val indexName = rs.getString("INDEX_NAME")
if (indexName == "${tableName}__KindPlusScheduledAtIndex") {
hasIndex = true
val columnName = rs.getString("COLUMN_NAME")
val ordinalPosition = rs.getShort("ORDINAL_POSITION")
// Check if 'id' is in position 3
if (columnName == "id" && ordinalPosition == 3.toShort()) {
hasIdColumn = true
}
}
}
}

// Run migration if index exists but doesn't have id column
hasIndex && !hasIdColumn
},
),
)
}
4 changes: 2 additions & 2 deletions ...vm/src/main/kotlin/org/funfix/delayedqueue/jvm/internals/jdbc/mssql/MsSqlServerAdapter.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,7 +112,7 @@ internal class MsSqlServerAdapter(driver: JdbcDriver, tableName: String) :
WITH (UPDLOCK, READPAST)
WHERE
[pKind] = ? AND [scheduledAt] <= ?
ORDER BY [scheduledAt]
ORDER BY [scheduledAt], [id]
"""

return conn.prepareStatement(sql) { stmt ->
Expand Down Expand Up @@ -153,7 +153,7 @@ internal class MsSqlServerAdapter(driver: JdbcDriver, tableName: String) :
WITH (UPDLOCK, READPAST)
WHERE
[pKind] = ? AND [scheduledAt] <= ?
ORDER BY [scheduledAt]
ORDER BY [scheduledAt], [id]
)
"""

Expand Down
38 changes: 36 additions & 2 deletions ...src/main/kotlin/org/funfix/delayedqueue/jvm/internals/jdbc/mssql/MsSqlServerMigrations.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,45 @@ internal object MsSqlServerMigrations {
ON [$tableName] ([pKind], [pKey]);

CREATE INDEX [${tableName}__KindPlusScheduledAtIndex]
ON [$tableName]([pKind], [scheduledAt]);
ON [$tableName]([pKind], [scheduledAt], [id]);

CREATE INDEX [${tableName}__LockUuidPlusIdIndex]
ON [$tableName]([lockUuid], [id]);
""",
)
),
// Migration 2: Update index to include 'id' column for deterministic ordering
Migration(
sql =
"""
IF EXISTS (SELECT * FROM sys.indexes WHERE name = '${tableName}__KindPlusScheduledAtIndex')
DROP INDEX [${tableName}__KindPlusScheduledAtIndex] ON [$tableName];
CREATE INDEX [${tableName}__KindPlusScheduledAtIndex]
ON [$tableName]([pKind], [scheduledAt], [id]);
""",
needsExecution = { conn ->
// Check if index exists but doesn't have the 'id' column
val metadata = conn.underlying.metaData
var hasIndex = false
var hasIdColumn = false

metadata.getIndexInfo(null, null, tableName, false, false).use { rs ->
while (rs.next()) {
val indexName = rs.getString("INDEX_NAME")
if (indexName == "${tableName}__KindPlusScheduledAtIndex") {
hasIndex = true
val columnName = rs.getString("COLUMN_NAME")
val ordinalPosition = rs.getShort("ORDINAL_POSITION")
// Check if 'id' is in position 3
if (columnName == "id" && ordinalPosition == 3.toShort()) {
hasIdColumn = true
}
}
}
}

// Run migration if index exists but doesn't have id column
hasIndex && !hasIdColumn
},
),
)
}
4 changes: 2 additions & 2 deletions ...ue-jvm/src/main/kotlin/org/funfix/delayedqueue/jvm/internals/jdbc/oracle/OracleAdapter.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,7 @@ internal class OracleAdapter(driver: JdbcDriver, tableName: String) :
SELECT ROWID
FROM "$tableName"
WHERE "pKind" = ? AND "scheduledAt" <= ?
ORDER BY "scheduledAt"
ORDER BY "scheduledAt", "id"
FETCH FIRST 1 ROWS ONLY
)
FOR UPDATE SKIP LOCKED
Expand Down Expand Up @@ -180,7 +180,7 @@ internal class OracleAdapter(driver: JdbcDriver, tableName: String) :
SELECT ROWID
FROM "$tableName"
WHERE "pKind" = ? AND "scheduledAt" <= ?
ORDER BY "scheduledAt"
ORDER BY "scheduledAt", "id"
FETCH FIRST $limit ROWS ONLY
)
FOR UPDATE SKIP LOCKED
Expand Down
45 changes: 43 additions & 2 deletions ...jvm/src/main/kotlin/org/funfix/delayedqueue/jvm/internals/jdbc/oracle/OracleMigrations.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,11 +31,52 @@ internal object OracleMigrations {
ON "$tableName"("pKey", "pKind");

CREATE INDEX "${tableName}__KindPlusScheduledAtIndex"
ON "$tableName"("pKind", "scheduledAt");
ON "$tableName"("pKind", "scheduledAt", "id");

CREATE INDEX "${tableName}__LockUuidPlusIdIndex"
ON "$tableName"("lockUuid", "id");
""",
)
),
// Migration 2: Update index to include 'id' column for deterministic ordering
Migration(
sql =
"""
BEGIN
EXECUTE IMMEDIATE 'DROP INDEX "${tableName}__KindPlusScheduledAtIndex"';
EXCEPTION
WHEN OTHERS THEN
IF SQLCODE != -1418 THEN
RAISE;
END IF;
END;
/
CREATE INDEX "${tableName}__KindPlusScheduledAtIndex"
ON "$tableName"("pKind", "scheduledAt", "id");
""",
needsExecution = { conn ->
// Check if index exists but doesn't have the 'id' column
val metadata = conn.underlying.metaData
var hasIndex = false
var hasIdColumn = false

metadata.getIndexInfo(null, null, tableName, false, false).use { rs ->
while (rs.next()) {
val indexName = rs.getString("INDEX_NAME")
if (indexName == "${tableName}__KindPlusScheduledAtIndex") {
hasIndex = true
val columnName = rs.getString("COLUMN_NAME")
val ordinalPosition = rs.getShort("ORDINAL_POSITION")
// Check if 'id' is in position 3
if (columnName == "id" && ordinalPosition == 3.toShort()) {
hasIdColumn = true
}
}
}
}

// Run migration if index exists but doesn't have id column
hasIndex && !hasIdColumn
},
),
)
}
4 changes: 2 additions & 2 deletions .../src/main/kotlin/org/funfix/delayedqueue/jvm/internals/jdbc/postgres/PostgreSQLAdapter.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,7 @@ internal class PostgreSQLAdapter(driver: JdbcDriver, tableName: String) :
"createdAt"
FROM "$tableName"
WHERE "pKind" = ? AND "scheduledAt" <= ?
ORDER BY "scheduledAt"
ORDER BY "scheduledAt", "id"
LIMIT 1
FOR UPDATE SKIP LOCKED
"""
Expand Down Expand Up @@ -140,7 +140,7 @@ internal class PostgreSQLAdapter(driver: JdbcDriver, tableName: String) :
FROM "$tableName"
WHERE
"pKind" = ? AND "scheduledAt" <= ?
ORDER BY "scheduledAt"
ORDER BY "scheduledAt", "id"
LIMIT $limit
FOR UPDATE SKIP LOCKED
)
Expand Down
37 changes: 35 additions & 2 deletions ...c/main/kotlin/org/funfix/delayedqueue/jvm/internals/jdbc/postgres/PostgreSQLMigrations.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,11 +31,44 @@ internal object PostgreSQLMigrations {
ON "$tableName"("pKey", "pKind");

CREATE INDEX "${tableName}__KindPlusScheduledAtIndex"
ON "$tableName"("pKind", "scheduledAt");
ON "$tableName"("pKind", "scheduledAt", "id");

CREATE INDEX "${tableName}__LockUuidPlusIdIndex"
ON "$tableName"("lockUuid", "id");
""",
)
),
// Migration 2: Update index to include 'id' column for deterministic ordering
Migration(
sql =
"""
DROP INDEX IF EXISTS "${tableName}__KindPlusScheduledAtIndex";
CREATE INDEX "${tableName}__KindPlusScheduledAtIndex"
ON "$tableName"("pKind", "scheduledAt", "id");
""",
needsExecution = { conn ->
// Check if index exists but doesn't have the 'id' column
val metadata = conn.underlying.metaData
var hasIndex = false
var hasIdColumn = false

metadata.getIndexInfo(null, null, tableName, false, false).use { rs ->
while (rs.next()) {
val indexName = rs.getString("INDEX_NAME")
if (indexName == "${tableName}__KindPlusScheduledAtIndex") {
hasIndex = true
val columnName = rs.getString("COLUMN_NAME")
val ordinalPosition = rs.getShort("ORDINAL_POSITION")
// Check if 'id' is in position 3
if (columnName == "id" && ordinalPosition == 3.toShort()) {
hasIdColumn = true
}
}
}
}

// Run migration if index exists but doesn't have id column
hasIndex && !hasIdColumn
},
),
)
}
Loading
Loading