Add support to set minimum time between checkpoints. (#413)

* Add support to set minimum time between checkpoints.

This is useful when uploading data on watermark or in hybrid mode.

* fix code smell

* Add check so initial checkpoint is always taken

* Add to flowtide builder

src/FlowtideDotNet.Base/Engine/DataflowStreamBuilder.cs

@@ -110,6 +110,12 @@ public DataflowStreamBuilder WaitForCheckpointAfterInitialData(bool wait)
             return this;
         }
 
+        public DataflowStreamBuilder SetMinimumTimeBetweenCheckpoint(TimeSpan timeSpan)
+        {
+            _dataflowStreamOptions.MinimumTimeBetweenCheckpoints = timeSpan;
+            return this;
+        }
+
         public DataflowStream Build()
         {
             if (_stateManagerOptions == null)

src/FlowtideDotNet.Base/Engine/Internal/DataflowStreamOptions.cs

@@ -21,5 +21,7 @@ namespace FlowtideDotNet.Base.Engine.Internal
     internal class DataflowStreamOptions
     {
         public bool WaitForCheckpointAfterInitialData { get; set; } = true;
+
+        public TimeSpan? MinimumTimeBetweenCheckpoints { get; set; }
     }
 }

src/FlowtideDotNet.Base/Engine/Internal/StateMachine/RunningStreamState.cs

@@ -128,6 +128,7 @@ private void CheckpointCompleted()
 
             lock (_context._checkpointLock)
             {
+                _context._initialCheckpointTaken = true;
                 if (_context.checkpointTask != null)
                 {
                     _context._scheduleCheckpointTask = null;

src/FlowtideDotNet.Base/Engine/Internal/StateMachine/StreamContext.cs

@@ -81,6 +81,8 @@ internal class StreamContext : IStreamTriggerCaller, IAsyncDisposable
 
         internal FlowtideDotNet.Storage.StateManager.StateManagerSync<StreamState> _stateManager;
         internal readonly ILogger<StreamContext> _logger;
+
+        internal bool _initialCheckpointTaken = false;
 
 
         public StreamContext(
@@ -336,6 +338,15 @@ internal void TryScheduleCheckpointIn(TimeSpan timeSpan)
         internal bool TryScheduleCheckpointIn_NoLock(TimeSpan timeSpan)
         {
             Debug.Assert(Monitor.IsEntered(_checkpointLock));
+
+            // Check if minimum time has been set, if so default it to the minimum time.
+            if (_dataflowStreamOptions.MinimumTimeBetweenCheckpoints != null &&
+                _initialCheckpointTaken &&
+                _dataflowStreamOptions.MinimumTimeBetweenCheckpoints.Value.CompareTo(timeSpan) > 0)
+            {
+                timeSpan = _dataflowStreamOptions.MinimumTimeBetweenCheckpoints.Value;
+            }
+
             var triggerTime = DateTime.Now.Add(timeSpan);
 
             // Check if a checkpoint is already running, if so, add that a checkpoint is waiting

src/FlowtideDotNet.Core/Engine/FlowtideBuilder.cs

@@ -110,6 +110,12 @@ public FlowtideBuilder SetGetTimestampUpdateInterval(TimeSpan interval)
             return this;
         }
 
+        public FlowtideBuilder SetMinimumTimeBetweenCheckpoint(TimeSpan timeSpan)
+        {
+            dataflowStreamBuilder.SetMinimumTimeBetweenCheckpoint(timeSpan);
+            return this;
+        }
+
         private string ComputePlanHash()
         {
             Debug.Assert(_plan != null, "Plan should not be null.");