Add durability-provider-specific support for implicit entity deletion. (#2251)

Author: sebastianburckhardt

src/WebJobs.Extensions.DurableTask/DurabilityProvider.cs

@@ -76,6 +76,11 @@ public DurabilityProvider(string storageProviderName, IOrchestrationService serv
         /// </summary>
         public virtual bool GuaranteesOrderedDelivery => false;
 
+        /// <summary>
+        /// Specifies whether this backend supports implicit deletion of entities.
+        /// </summary>
+        public virtual bool SupportsImplicitEntityDeletion => false;
+
         /// <summary>
         /// JSON representation of configuration to emit in telemetry.
         /// </summary>

src/WebJobs.Extensions.DurableTask/DurableTaskExtension.cs

@@ -258,6 +258,8 @@ public string HubName
         internal TimeSpan MessageReorderWindow
             => this.defaultDurabilityProvider.GuaranteesOrderedDelivery ? TimeSpan.Zero : TimeSpan.FromMinutes(this.Options.EntityMessageReorderWindowInMinutes);
 
+        internal bool UseImplicitEntityDeletion => this.defaultDurabilityProvider.SupportsImplicitEntityDeletion;
+
         internal IApplicationLifetimeWrapper HostLifetimeService { get; } = HostLifecycleService.NoOp;
 
         internal OutOfProcOrchestrationProtocol OutOfProcProtocol { get; }

src/WebJobs.Extensions.DurableTask/Listener/TaskEntityShim.cs

@@ -147,7 +147,7 @@ internal void Rehydrate(string serializedInput)
         {
             if (serializedInput == null)
             {
-                // this instance was automatically started by DTFx
+                // this instance was automatically started by DTFx, or a previous execution called continueAsNew(null)
                 this.context.State = new SchedulerState();
             }
             else
@@ -247,10 +247,18 @@ public override async Task<string> Execute(OrchestrationContext innerContext, st
                         this.context.State.Suspended = true;
                     }
 
-                    var jstate = JToken.FromObject(this.context.State);
-
-                    // continue as new
-                    innerContext.ContinueAsNew(jstate);
+                    if (this.Config.UseImplicitEntityDeletion && this.context.State.IsEmpty)
+                    {
+                        // this entity scheduler is idle and the entity is deleted, so the instance and history can be removed from storage
+                        // we convey this to the durability provider by issuing a continue-as-new with null input
+                        innerContext.ContinueAsNew(null);
+                    }
+                    else
+                    {
+                        // we persist the state of the entity scheduler and entity by issuing a continue-as-new
+                        var jstate = JToken.FromObject(this.context.State);
+                        innerContext.ContinueAsNew(jstate);
+                    }
                 }
             }
             catch (Exception e)