Merge pull request #14 from raisedapp/develop

felixclase · web-flow · commit 361a97ffc6a3 · 2019-12-20T09:37:51.000-04:00
Version 0.2.0
diff --git a/src/main/Hangfire.Storage.SQLite/CountersAggregator.cs b/src/main/Hangfire.Storage.SQLite/CountersAggregator.cs
@@ -94,7 +94,7 @@ public void Execute(CancellationToken cancellationToken)
 
                             foreach (var counter in aggregatedCounters)
                             {
-                                counter.Value = counter.Value + item.Value;
+                                counter.Value += item.Value;
                                 counter.ExpireAt = item.ExpireAt > aggregatedItem.ExpireAt
                                     ? (item.ExpireAt > DateTime.MinValue ? item.ExpireAt : DateTime.MinValue)
                                     : (aggregatedItem.ExpireAt > DateTime.MinValue ? 
diff --git a/src/main/Hangfire.Storage.SQLite/Entities/AggregatedCounter.cs b/src/main/Hangfire.Storage.SQLite/Entities/AggregatedCounter.cs
@@ -5,6 +5,7 @@
 
 namespace Hangfire.Storage.SQLite.Entities
 {
+    [Table("AggregatedCounter")]
     public class AggregatedCounter
     {
         [PrimaryKey]
diff --git a/src/main/Hangfire.Storage.SQLite/Entities/Counter.cs b/src/main/Hangfire.Storage.SQLite/Entities/Counter.cs
@@ -5,6 +5,7 @@
 
 namespace Hangfire.Storage.SQLite.Entities
 {
+    [Table(DefaultValues.CounterTblName)]
     public class Counter
     {
         [PrimaryKey]
diff --git a/src/main/Hangfire.Storage.SQLite/Entities/DefaultValues.cs b/src/main/Hangfire.Storage.SQLite/Entities/DefaultValues.cs
@@ -23,5 +23,25 @@ internal static class DefaultValues
         public const int MaxLengthReasonColumn = 100;
 
         public const int MaxLengthVarCharColumn = 16000;
+
+        public const string AggregatedCounterTblName = "AggregatedCounter";
+
+        public const string CounterTblName = "Counter";
+
+        public const string DistributedLockTblName = "DistributedLock";
+
+        public const string HangfireJobTblName = "Job";
+
+        public const string HangfireListTblName = "List";
+
+        public const string HangfireServerTblName = "Server";
+
+        public const string HashTblName = "Hash";
+
+        public const string JobParameterTblName = "JobParameter";
+
+        public const string SetTblName = "Set";
+
+        public const string StateTblName = "State";
     }
 }
diff --git a/src/main/Hangfire.Storage.SQLite/Entities/DistributedLock.cs b/src/main/Hangfire.Storage.SQLite/Entities/DistributedLock.cs
@@ -8,6 +8,7 @@ namespace Hangfire.Storage.SQLite.Entities
     /// <summary>
     /// Document used for holding a distributed lock in SQLite.
     /// </summary>
+    [Table(DefaultValues.DistributedLockTblName)]
     public class DistributedLock
     {
         /// <summary>
diff --git a/src/main/Hangfire.Storage.SQLite/Entities/HangfireJob.cs b/src/main/Hangfire.Storage.SQLite/Entities/HangfireJob.cs
@@ -5,7 +5,7 @@
 
 namespace Hangfire.Storage.SQLite.Entities
 {
-    [Table("Job")]
+    [Table(DefaultValues.HangfireJobTblName)]
     public class HangfireJob
     {
         [PrimaryKey, AutoIncrement]
diff --git a/src/main/Hangfire.Storage.SQLite/Entities/HangfireList.cs b/src/main/Hangfire.Storage.SQLite/Entities/HangfireList.cs
@@ -5,7 +5,7 @@
 
 namespace Hangfire.Storage.SQLite.Entities
 {
-    [Table("List")]
+    [Table(DefaultValues.HangfireListTblName)]
     public class HangfireList
     {
         [PrimaryKey, AutoIncrement]
diff --git a/src/main/Hangfire.Storage.SQLite/Entities/HangfireServer.cs b/src/main/Hangfire.Storage.SQLite/Entities/HangfireServer.cs
@@ -5,7 +5,7 @@
 
 namespace Hangfire.Storage.SQLite.Entities
 {
-    [Table("Server")]
+    [Table(DefaultValues.HangfireServerTblName)]
     public class HangfireServer
     {
         [PrimaryKey]
diff --git a/src/main/Hangfire.Storage.SQLite/Entities/Hash.cs b/src/main/Hangfire.Storage.SQLite/Entities/Hash.cs
@@ -5,6 +5,7 @@
 
 namespace Hangfire.Storage.SQLite.Entities
 {
+    [Table(DefaultValues.HashTblName)]
     public class Hash
     {
         [PrimaryKey, AutoIncrement]
diff --git a/src/main/Hangfire.Storage.SQLite/Entities/JobParameter.cs b/src/main/Hangfire.Storage.SQLite/Entities/JobParameter.cs
@@ -5,6 +5,7 @@
 
 namespace Hangfire.Storage.SQLite.Entities
 {
+    [Table(DefaultValues.JobParameterTblName)]
     public class JobParameter
     {
         [PrimaryKey, AutoIncrement]
diff --git a/src/main/Hangfire.Storage.SQLite/Entities/JobQueue.cs b/src/main/Hangfire.Storage.SQLite/Entities/JobQueue.cs
@@ -5,6 +5,7 @@
 
 namespace Hangfire.Storage.SQLite.Entities
 {
+    [Table("JobQueue")]
     public class JobQueue
     {
         [PrimaryKey, AutoIncrement]
diff --git a/src/main/Hangfire.Storage.SQLite/Entities/Set.cs b/src/main/Hangfire.Storage.SQLite/Entities/Set.cs
@@ -5,6 +5,7 @@
 
 namespace Hangfire.Storage.SQLite.Entities
 {
+    [Table(DefaultValues.SetTblName)]
     public class Set
     {
         [PrimaryKey, AutoIncrement]
diff --git a/src/main/Hangfire.Storage.SQLite/Entities/State.cs b/src/main/Hangfire.Storage.SQLite/Entities/State.cs
@@ -5,6 +5,7 @@
 
 namespace Hangfire.Storage.SQLite.Entities
 {
+    [Table(DefaultValues.StateTblName)]
     public class State
     {
         [PrimaryKey, AutoIncrement]
diff --git a/src/main/Hangfire.Storage.SQLite/ExpirationManager.cs b/src/main/Hangfire.Storage.SQLite/ExpirationManager.cs
@@ -1,6 +1,7 @@
 ﻿using Hangfire.Annotations;
 using Hangfire.Logging;
 using Hangfire.Server;
+using Hangfire.Storage.SQLite.Entities;
 using System;
 using System.Collections.Generic;
 using System.Text;
@@ -11,10 +12,30 @@ namespace Hangfire.Storage.SQLite
     /// <summary>
     /// Represents Hangfire expiration manager for LiteDB database
     /// </summary>
-    #pragma warning disable CS0618
+#pragma warning disable CS0618
     public class ExpirationManager : IBackgroundProcess, IServerComponent
-    #pragma warning restore CS0618 
+#pragma warning restore CS0618
     {
+        private const string DistributedLockKey = "locks:expirationmanager";
+        private static readonly TimeSpan DefaultLockTimeout = TimeSpan.FromMinutes(5);
+
+        // This value should be high enough to optimize the deletion as much, as possible,
+        // reducing the number of queries. But low enough to cause lock escalations (it
+        // appears, when ~5000 locks were taken, but this number is a subject of version).
+        // Note, that lock escalation may also happen during the cascade deletions for
+        // State (3-5 rows/job usually) and JobParameters (2-3 rows/job usually) tables.
+        private const int NumberOfRecordsInSinglePass = 1000;
+
+        private static readonly string[] ProcessedTables =
+        {
+            DefaultValues.AggregatedCounterTblName,
+            DefaultValues.CounterTblName,
+            DefaultValues.HangfireJobTblName,
+            DefaultValues.HangfireListTblName,
+            DefaultValues.SetTblName,
+            DefaultValues.HashTblName,
+        };
+
         private static readonly ILog Logger = LogProvider.For<ExpirationManager>();
 
         private readonly SQLiteStorage _storage;
@@ -56,16 +77,54 @@ public void Execute([NotNull] BackgroundProcessContext context)
         public void Execute(CancellationToken cancellationToken)
         {
             HangfireDbContext connection = _storage.CreateAndOpenConnection();
-            DateTime now = DateTime.UtcNow;
 
-            connection.HangfireJobRepository.Delete(_ => _.ExpireAt > DateTime.MinValue && _.ExpireAt < now);
-            connection.AggregatedCounterRepository.Delete(_ => _.ExpireAt > DateTime.MinValue && _.ExpireAt < now);
-            connection.CounterRepository.Delete(_ => _.ExpireAt > DateTime.MinValue && _.ExpireAt < now);
-            connection.HashRepository.Delete(_ => _.ExpireAt > DateTime.MinValue && _.ExpireAt < now);
-            connection.SetRepository.Delete(_ => _.ExpireAt > DateTime.MinValue && _.ExpireAt < now);
-            connection.HangfireListRepository.Delete(_ => _.ExpireAt > DateTime.MinValue  && _.ExpireAt < now);
+            foreach (var table in ProcessedTables)
+            {
+                Logger.Debug($"Removing outdated records from the '{table}' table...");
+
+                int affected;
+                do
+                {
+                    affected = RemoveExpireRows(connection, table);
+                } while (affected == NumberOfRecordsInSinglePass);
+
+                Logger.Trace($"Outdated records removed from the '{table}' table...");
+            }
 
             cancellationToken.WaitHandle.WaitOne(_checkInterval);
         }
+
+        private int RemoveExpireRows(HangfireDbContext db, string table) 
+        {
+            var now = DateTime.UtcNow;
+            var deleteScript = $"DELETE FROM [{table}] WHERE rowid IN (SELECT rowid FROM [{table}] WHERE ExpireAt > {DateTime.MinValue.Ticks} AND ExpireAt < {now.Ticks} LIMIT {NumberOfRecordsInSinglePass})";
+            int rowsAffected = 0;
+
+            try
+            {
+                var _lock = new SQLiteDistributedLock(DistributedLockKey, DefaultLockTimeout,
+                    db, db.StorageOptions);
+
+                using (_lock) 
+                {
+                    rowsAffected = db.Database.Execute(deleteScript);
+                }
+            }
+            catch (DistributedLockTimeoutException e) when (e.Resource == DistributedLockKey)
+            {
+                // DistributedLockTimeoutException here doesn't mean that outdated records weren't removed.
+                // It just means another Hangfire server did this work.
+                Logger.Log(
+                    LogLevel.Debug,
+                    () => $@"An exception was thrown during acquiring distributed lock on the {DistributedLockKey} resource within {DefaultLockTimeout.TotalSeconds} seconds. Outdated records were not removed. It will be retried in {_checkInterval.TotalSeconds} seconds.",
+                    e);
+            }
+            catch (Exception e)
+            {
+                Logger.Log(LogLevel.Error, () => $"Error in RemoveExpireRows Method. Detail: {e.ToString()}", e);
+            }
+
+            return rowsAffected;
+        }
     }
 }
diff --git a/src/main/Hangfire.Storage.SQLite/Hangfire.Storage.SQLite.csproj b/src/main/Hangfire.Storage.SQLite/Hangfire.Storage.SQLite.csproj
@@ -7,7 +7,7 @@
     <TargetFrameworks>netstandard2.0;net45</TargetFrameworks>
   </PropertyGroup>
   <PropertyGroup>
-    <Version>0.1.0</Version>
+    <Version>0.2.0</Version>
     <Authors>RaisedApp</Authors>
     <Company>RaisedApp</Company>
     <Copyright>Copyright © 2019 - Present</Copyright>
@@ -20,9 +20,9 @@
     <title>Hangfire Storage SQLite</title>    
     <Description>An Alternative SQLite Storage for Hangfire</Description>
     <PackageReleaseNotes>
-      0.1.0
-      - Fix Recurring Job Count (Dashboard)
-      - Fix Dequeue method
+      0.2.0
+      - Distributed Lock In ExpirationManager
+      - ExpireAt In JobDetails Method
     </PackageReleaseNotes>
   </PropertyGroup>
   <ItemGroup>
@@ -36,4 +36,4 @@
     <PackageReference Include="sqlite-net-pcl" Version="1.6.292" />
   </ItemGroup>
 
-</Project>
+</Project>
diff --git a/src/main/Hangfire.Storage.SQLite/SQLiteMonitoringApi.cs b/src/main/Hangfire.Storage.SQLite/SQLiteMonitoringApi.cs
@@ -481,7 +481,8 @@ public JobDetailsDto JobDetails(string jobId)
                     CreatedAt = DateTime.UtcNow,
                     Job = DeserializeJob(job.InvocationData, job.Arguments),
                     History = history,
-                    Properties = jobParameters
+                    Properties = jobParameters,
+                    ExpireAt = job.ExpireAt
                 };
             });
         }
diff --git a/src/test/Hangfire.Storage.SQLite.Test/SQLiteDistributedLockFacts.cs b/src/test/Hangfire.Storage.SQLite.Test/SQLiteDistributedLockFacts.cs
@@ -165,7 +165,7 @@ public void Ctor_WaitForLock_OnlySingleLockCanBeAcquired()
 
             manualResetEvent.Set();
 
-            threads.ForEach(t => Assert.True(t.Join(TimeSpan.FromMinutes(1)), "Thread is hanging unexpected"));
+            threads.ForEach(t => Assert.True(t.Join(TimeSpan.FromSeconds(90)), "Thread is hanging unexpected"));
 
             // All the threads should report success.
             Interlocked.MemoryBarrier();

Original file line number	Diff line number	Diff line change
`@@ -94,7 +94,7 @@ public void Execute(CancellationToken cancellationToken)`
`94`	`94`
`95`	`95`	`foreach (var counter in aggregatedCounters)`
`96`	`96`	`{`
`97`		`- counter.Value = counter.Value + item.Value;`
	`97`	`+ counter.Value += item.Value;`
`98`	`98`	`counter.ExpireAt = item.ExpireAt > aggregatedItem.ExpireAt`
`99`	`99`	`? (item.ExpireAt > DateTime.MinValue ? item.ExpireAt : DateTime.MinValue)`
`100`	`100`	`: (aggregatedItem.ExpireAt > DateTime.MinValue ?`
Original file line number	Diff line number	Diff line change
`@@ -5,6 +5,7 @@`
`5`	`5`
`6`	`6`	`namespace Hangfire.Storage.SQLite.Entities`
`7`	`7`	`{`
	`8`	`+ [Table("AggregatedCounter")]`
`8`	`9`	`public class AggregatedCounter`
`9`	`10`	`{`
`10`	`11`	`[PrimaryKey]`
Original file line number	Diff line number	Diff line change
`@@ -5,6 +5,7 @@`
`5`	`5`
`6`	`6`	`namespace Hangfire.Storage.SQLite.Entities`
`7`	`7`	`{`
	`8`	`+ [Table(DefaultValues.CounterTblName)]`
`8`	`9`	`public class Counter`
`9`	`10`	`{`
`10`	`11`	`[PrimaryKey]`
Original file line number	Diff line number	Diff line change
`@@ -8,6 +8,7 @@ namespace Hangfire.Storage.SQLite.Entities`
`8`	`8`	`/// <summary>`
`9`	`9`	`/// Document used for holding a distributed lock in SQLite.`
`10`	`10`	`/// </summary>`
	`11`	`+ [Table(DefaultValues.DistributedLockTblName)]`
`11`	`12`	`public class DistributedLock`
`12`	`13`	`{`
`13`	`14`	`/// <summary>`
Original file line number	Diff line number	Diff line change
`@@ -5,7 +5,7 @@`
`5`	`5`
`6`	`6`	`namespace Hangfire.Storage.SQLite.Entities`
`7`	`7`	`{`
`8`		`- [Table("Job")]`
	`8`	`+ [Table(DefaultValues.HangfireJobTblName)]`
`9`	`9`	`public class HangfireJob`
`10`	`10`	`{`
`11`	`11`	`[PrimaryKey, AutoIncrement]`
Original file line number	Diff line number	Diff line change
`@@ -5,7 +5,7 @@`
`5`	`5`
`6`	`6`	`namespace Hangfire.Storage.SQLite.Entities`
`7`	`7`	`{`
`8`		`- [Table("List")]`
	`8`	`+ [Table(DefaultValues.HangfireListTblName)]`
`9`	`9`	`public class HangfireList`
`10`	`10`	`{`
`11`	`11`	`[PrimaryKey, AutoIncrement]`
Original file line number	Diff line number	Diff line change
`@@ -5,7 +5,7 @@`
`5`	`5`
`6`	`6`	`namespace Hangfire.Storage.SQLite.Entities`
`7`	`7`	`{`
`8`		`- [Table("Server")]`
	`8`	`+ [Table(DefaultValues.HangfireServerTblName)]`
`9`	`9`	`public class HangfireServer`
`10`	`10`	`{`
`11`	`11`	`[PrimaryKey]`
Original file line number	Diff line number	Diff line change
`@@ -5,6 +5,7 @@`
`5`	`5`
`6`	`6`	`namespace Hangfire.Storage.SQLite.Entities`
`7`	`7`	`{`
	`8`	`+ [Table(DefaultValues.HashTblName)]`
`8`	`9`	`public class Hash`
`9`	`10`	`{`
`10`	`11`	`[PrimaryKey, AutoIncrement]`
Original file line number	Diff line number	Diff line change
`@@ -5,6 +5,7 @@`
`5`	`5`
`6`	`6`	`namespace Hangfire.Storage.SQLite.Entities`
`7`	`7`	`{`
	`8`	`+ [Table(DefaultValues.JobParameterTblName)]`
`8`	`9`	`public class JobParameter`
`9`	`10`	`{`
`10`	`11`	`[PrimaryKey, AutoIncrement]`