From 51ea48b63d3e848e8d9bd2dc3d8cff601cc47a7b Mon Sep 17 00:00:00 2001 From: Nicolaj Hartmann Date: Mon, 29 Dec 2025 14:28:56 +0100 Subject: [PATCH] Fix duplicate key error in FetchedJobs when job has multiple queue entries Add DISTINCT/DISTINCT ON to SQL queries in monitoring API to handle cases where a job has multiple entries in the jobqueue table. This prevents ArgumentException when the dashboard displays fetched jobs. --- .../PostgreSqlJobQueueMonitoringApi.cs | 6 +-- .../PostgreSqlMonitoringApi.cs | 12 +++--- .../PostgreSqlMonitoringApiFacts.cs | 43 +++++++++++++++++++ 3 files changed, 53 insertions(+), 8 deletions(-) diff --git a/src/Hangfire.PostgreSql/PostgreSqlJobQueueMonitoringApi.cs b/src/Hangfire.PostgreSql/PostgreSqlJobQueueMonitoringApi.cs index 4ec1be6b..7907638a 100644 --- a/src/Hangfire.PostgreSql/PostgreSqlJobQueueMonitoringApi.cs +++ b/src/Hangfire.PostgreSql/PostgreSqlJobQueueMonitoringApi.cs @@ -80,13 +80,13 @@ SELECT COUNT(*) private IEnumerable GetQueuedOrFetchedJobIds(string queue, bool fetched, int from, int perPage) { string sqlQuery = $@" - SELECT j.""id"" + SELECT DISTINCT j.""id"" FROM ""{_storage.Options.SchemaName}"".""jobqueue"" jq LEFT JOIN ""{_storage.Options.SchemaName}"".""job"" j ON jq.""jobid"" = j.""id"" - WHERE jq.""queue"" = @Queue + WHERE jq.""queue"" = @Queue AND jq.""fetchedat"" {(fetched ? "IS NOT NULL" : "IS NULL")} AND j.""id"" IS NOT NULL - ORDER BY jq.""fetchedat"", jq.""jobid"" + ORDER BY j.""id"" LIMIT @Limit OFFSET @Offset; "; diff --git a/src/Hangfire.PostgreSql/PostgreSqlMonitoringApi.cs b/src/Hangfire.PostgreSql/PostgreSqlMonitoringApi.cs index 63a5e4c6..145df1dc 100644 --- a/src/Hangfire.PostgreSql/PostgreSqlMonitoringApi.cs +++ b/src/Hangfire.PostgreSql/PostgreSqlMonitoringApi.cs @@ -433,13 +433,14 @@ private IPersistentJobQueueMonitoringApi GetQueueApi(string queueName) private JobList EnqueuedJobs(IEnumerable jobIds) { string enqueuedJobsSql = $@" - SELECT ""j"".""id"" ""Id"", ""j"".""invocationdata"" ""InvocationData"", ""j"".""arguments"" ""Arguments"", ""j"".""createdat"" ""CreatedAt"", + SELECT DISTINCT ON (""j"".""id"") ""j"".""id"" ""Id"", ""j"".""invocationdata"" ""InvocationData"", ""j"".""arguments"" ""Arguments"", ""j"".""createdat"" ""CreatedAt"", ""j"".""expireat"" ""ExpireAt"", ""s"".""name"" ""StateName"", ""s"".""reason"" ""StateReason"", ""s"".""data"" ""StateData"" FROM ""{_storage.Options.SchemaName}"".""job"" ""j"" LEFT JOIN ""{_storage.Options.SchemaName}"".""state"" ""s"" ON ""s"".""id"" = ""j"".""stateid"" LEFT JOIN ""{_storage.Options.SchemaName}"".""jobqueue"" ""jq"" ON ""jq"".""jobid"" = ""j"".""id"" WHERE ""j"".""id"" = ANY (@JobIds) - AND ""jq"".""fetchedat"" IS NULL; + AND ""jq"".""fetchedat"" IS NULL + ORDER BY ""j"".""id""; "; List jobs = UseConnection(connection => connection.Query(enqueuedJobsSql, @@ -528,14 +529,15 @@ private JobList FetchedJobs( IEnumerable jobIds) { string fetchedJobsSql = $@" - SELECT ""j"".""id"" ""Id"", ""j"".""invocationdata"" ""InvocationData"", ""j"".""arguments"" ""Arguments"", - ""j"".""createdat"" ""CreatedAt"", ""j"".""expireat"" ""ExpireAt"", ""jq"".""fetchedat"" ""FetchedAt"", + SELECT DISTINCT ON (""j"".""id"") ""j"".""id"" ""Id"", ""j"".""invocationdata"" ""InvocationData"", ""j"".""arguments"" ""Arguments"", + ""j"".""createdat"" ""CreatedAt"", ""j"".""expireat"" ""ExpireAt"", ""jq"".""fetchedat"" ""FetchedAt"", ""j"".""statename"" ""StateName"", ""s"".""reason"" ""StateReason"", ""s"".""data"" ""StateData"" FROM ""{_storage.Options.SchemaName}"".""job"" ""j"" LEFT JOIN ""{_storage.Options.SchemaName}"".""state"" ""s"" ON ""j"".""stateid"" = ""s"".""id"" LEFT JOIN ""{_storage.Options.SchemaName}"".""jobqueue"" ""jq"" ON ""jq"".""jobid"" = ""j"".""id"" WHERE ""j"".""id"" = ANY (@JobIds) - AND ""jq"".""fetchedat"" IS NOT NULL; + AND ""jq"".""fetchedat"" IS NOT NULL + ORDER BY ""j"".""id"", ""jq"".""fetchedat"" DESC; "; List jobs = UseConnection(connection => connection.Query(fetchedJobsSql, diff --git a/tests/Hangfire.PostgreSql.Tests/PostgreSqlMonitoringApiFacts.cs b/tests/Hangfire.PostgreSql.Tests/PostgreSqlMonitoringApiFacts.cs index 4f6f217a..4978a9fd 100644 --- a/tests/Hangfire.PostgreSql.Tests/PostgreSqlMonitoringApiFacts.cs +++ b/tests/Hangfire.PostgreSql.Tests/PostgreSqlMonitoringApiFacts.cs @@ -103,6 +103,49 @@ private void Commit( transaction.Commit(); } + [Fact] + [CleanDatabase] + public void FetchedJobs_WithDuplicateJobQueueEntries_DoesNotThrow() + { + string schemaName = ConnectionUtils.GetSchemaName(); + + string createJobSql = $@" + INSERT INTO ""{schemaName}"".""job"" (""invocationdata"", ""arguments"", ""createdat"") + VALUES (@InvocationData, @Arguments, NOW()) RETURNING ""id"""; + + string createJobQueueSql = $@" + INSERT INTO ""{schemaName}"".""jobqueue"" (""jobid"", ""queue"", ""fetchedat"") + VALUES (@JobId, @Queue, @FetchedAt)"; + + UseConnection(connection => { + Job job = Job.FromExpression(() => SampleMethod("test")); + InvocationData invocationData = InvocationData.SerializeJob(job); + + long jobId = connection.QuerySingle(createJobSql, + new { + InvocationData = new JsonParameter(SerializationHelper.Serialize(invocationData)), + Arguments = new JsonParameter(invocationData.Arguments, JsonParameter.ValueType.Array), + }); + + DateTime fetchedAt = DateTime.UtcNow; + connection.Execute(createJobQueueSql, new { JobId = jobId, Queue = "default", FetchedAt = fetchedAt }); + connection.Execute(createJobQueueSql, new { JobId = jobId, Queue = "default", FetchedAt = fetchedAt.AddSeconds(1) }); + + PostgreSqlStorage storage = _fixture.SafeInit(); + PostgreSqlStorageOptions options = new() { SchemaName = schemaName }; + + PostgreSqlJobQueueProvider provider = new(storage, options); + PersistentJobQueueProviderCollection providers = new(provider); + storage.QueueProviders = providers; + + IMonitoringApi monitoringApi = storage.GetMonitoringApi(); + JobList fetchedJobs = monitoringApi.FetchedJobs("default", 0, 10); + + Assert.NotNull(fetchedJobs); + Assert.Single(fetchedJobs); + }); + } + #pragma warning disable xUnit1013 // Public method should be marked as test public static void SampleMethod(string arg) #pragma warning restore xUnit1013 // Public method should be marked as test