Merge pull request #203 from Plasma/feature-expiration-manager-improvements

Improve ExpirationManager performance and options

Author: frankhommers

src/Hangfire.PostgreSql/ExpirationManager.cs

@@ -32,7 +32,6 @@ namespace Hangfire.PostgreSql
     internal class ExpirationManager : IBackgroundProcess, IServerComponent
     {
         private static readonly TimeSpan DelayBetweenPasses = TimeSpan.FromSeconds(1);
-        private const int NumberOfRecordsInSinglePass = 1000;
 
         private static readonly ILog Logger = LogProvider.GetLogger(typeof(ExpirationManager));
 
@@ -56,7 +55,7 @@ internal class ExpirationManager : IBackgroundProcess, IServerComponent
         private readonly TimeSpan _checkInterval;
 
         public ExpirationManager(PostgreSqlStorage storage, PostgreSqlStorageOptions options)
-            : this(storage, options, TimeSpan.FromHours(1))
+            : this(storage, options, options.JobExpirationCheckInterval)
         {
         }
 
@@ -93,7 +92,7 @@ DELETE FROM """ + _options.SchemaName + @""".""{0}""
     FROM """ + _options.SchemaName + @""".""{0}"" 
     WHERE ""expireat"" < NOW() AT TIME ZONE 'UTC' 
     LIMIT {1}
-)", table, NumberOfRecordsInSinglePass.ToString(CultureInfo.InvariantCulture)), transaction);
+)", table, _options.DeleteExpiredBatchSize.ToString(CultureInfo.InvariantCulture)), transaction);
 
                             transaction.Commit();
                         }

src/Hangfire.PostgreSql/PostgreSqlStorageOptions.cs

@@ -29,16 +29,20 @@ public class PostgreSqlStorageOptions
         private TimeSpan _invisibilityTimeout;
         private TimeSpan _distributedLockTimeout;
         private TimeSpan _transactionSerializationTimeout;
+        private TimeSpan _jobExpirationCheckInterval;
+        private int _deleteExpiredBatchSize;
 
         public PostgreSqlStorageOptions()
         {
             QueuePollInterval = TimeSpan.FromSeconds(15);
             InvisibilityTimeout = TimeSpan.FromMinutes(30);
             DistributedLockTimeout = TimeSpan.FromMinutes(10);
 	        TransactionSynchronisationTimeout = TimeSpan.FromMilliseconds(500);
+	        JobExpirationCheckInterval = TimeSpan.FromHours(1);
             SchemaName = "hangfire";
             UseNativeDatabaseTransactions = true;
             PrepareSchemaIfNecessary = true;
+            DeleteExpiredBatchSize = 1000;
         }
 
         public TimeSpan QueuePollInterval
@@ -81,7 +85,30 @@ public TimeSpan TransactionSynchronisationTimeout
 		    }
 		}
 
-        public bool UseNativeDatabaseTransactions { get; set; }
+		public TimeSpan JobExpirationCheckInterval
+		{
+			get => _jobExpirationCheckInterval;
+			set
+			{
+				ThrowIfValueIsNotPositive(value, nameof(JobExpirationCheckInterval));
+				_jobExpirationCheckInterval = value;
+			}
+		}
+
+		/// <summary>
+		/// Gets or sets the number of records deleted in a single batch in expiration manager
+		/// </summary>
+		public int DeleteExpiredBatchSize
+		{
+			get => _deleteExpiredBatchSize;
+			set 
+			{
+				ThrowIfValueIsNotPositive(value, nameof(DeleteExpiredBatchSize));
+                _deleteExpiredBatchSize = value;
+			}
+		}
+
+		public bool UseNativeDatabaseTransactions { get; set; }
         public bool PrepareSchemaIfNecessary { get; set; }
         public string SchemaName { get; set; }
         public bool EnableTransactionScopeEnlistment { get; set; }
@@ -99,5 +126,11 @@ private static void ThrowIfValueIsNotPositive(TimeSpan value, string fieldName)
                 throw new ArgumentException(message, nameof(value));
             }
         }
+
+        private static void ThrowIfValueIsNotPositive(int value, string fieldName)
+        {
+            if (value <= 0)
+		        throw new ArgumentException($"The {fieldName} property value should be positive. Given: {value}.");
+        }
     }
 }

src/Hangfire.PostgreSql/Scripts/Install.v15.sql

@@ -0,0 +1,17 @@
+SET search_path = 'hangfire';
+
+DO
+$$
+BEGIN
+  IF EXISTS (SELECT 1 FROM "schema" WHERE "version"::integer >= 15) THEN
+    RAISE EXCEPTION 'version-already-applied';
+  END IF;
+END
+$$;
+
+CREATE INDEX ix_hangfire_job_expireat ON "job" (expireat);
+CREATE INDEX ix_hangfire_list_expireat ON "list" (expireat);
+CREATE INDEX ix_hangfire_set_expireat ON "set" (expireat);
+CREATE INDEX ix_hangfire_hash_expireat ON "hash" (expireat);
+
+RESET search_path;

tests/Hangfire.PostgreSql.Tests/PostgreSqlInstallerFacts.cs

@@ -20,7 +20,7 @@ public void InstallingSchemaUpdatesVersionAndShouldNotThrowAnException()
 					PostgreSqlObjectsInstaller.Install(connection, schemaName);
 
 					var lastVersion = connection.Query<int>(@"select version from """ + schemaName + @""".""schema""").Single();
-					Assert.Equal(14, lastVersion);
+					Assert.Equal(15, lastVersion);
 
 					connection.Execute($@"DROP SCHEMA ""{schemaName}"" CASCADE;");
 				});