fix(raft): correct ambiguous-write classification, prevent reclaimed-job loss, add operational logging

src/Hangfire.Raft/Cluster/HangfireStateMachine.cs

@@ -125,6 +125,11 @@ protected override async ValueTask RestoreAsync(FileInfo snapshotFile, Cancellat
             var snapshot = await File.ReadAllBytesAsync(snapshotFile.FullName, token).ConfigureAwait(false);
             using var reader = new BinaryReader(new MemoryStream(snapshot, writable: false), Encoding.UTF8);
             Store.LoadSnapshot(reader);
+
+            // A restore-from-snapshot boot (or a follower installing a leader-pushed snapshot) is otherwise
+            // indistinguishable in the logs from a fresh empty start; record it so an operator can confirm
+            // how much state the node came up on.
+            _logger.LogInformation("Restored state-machine snapshot from {File} ({Bytes} bytes).", snapshotFile.FullName, snapshot.Length);
         }
         catch (OperationCanceledException) when (token.IsCancellationRequested)
         {

src/Hangfire.Raft/Cluster/RaftStorageCluster.cs

@@ -147,6 +147,13 @@ public static async Task<RaftStorageCluster> StartAsync(RaftStorageOptions optio
         var cluster = new RaftStorageCluster(options);
         try
         {
+            // Make the boot decision observable: an operator needs to tell a brand-new cold start apart from
+            // a resume-from-disk, especially because a deleted config file with a surviving WAL is misread as
+            // a first start (see the _coldStart comment) and silently re-seeds a one-member cluster.
+            cluster._logger.LogInformation(
+                "Starting Raft node {Self}: coldStart={ColdStart}, configuredMembers={Members}, walPath={WalPath}.",
+                options.Self, cluster._coldStart, cluster._clusterMembers.Count, options.WalPath);
+
             // Restore the latest snapshot into the state machine before building the WAL over it, so the
             // tail replay below applies on top of the restored state rather than an empty store. This is
             // the strictly-necessary explicit step: RaftCluster.StartAsync replays the committed tail (via
@@ -171,6 +178,10 @@ public static async Task<RaftStorageCluster> StartAsync(RaftStorageOptions optio
             if (hasExistingLog)
                 await cluster._wal.InitializeAsync(token).ConfigureAwait(false);
 
+            cluster._logger.LogInformation(
+                "Raft node {Self} restored local state: existingLog={HasLog}, appliedIndex={Applied}, committedIndex={Committed}.",
+                options.Self, hasExistingLog, cluster._wal.LastAppliedIndex, cluster._wal.LastCommittedEntryIndex);
+
             cluster._cluster = new RaftCluster(cluster._configuration) { AuditTrail = cluster._wal };
 
             // Seed a multi-node cluster's committed membership on first start so it forms regardless of
@@ -181,6 +192,7 @@ public static async Task<RaftStorageCluster> StartAsync(RaftStorageOptions optio
             cluster._forwardingServer.Start();
             await cluster._cluster.StartAsync(token).ConfigureAwait(false);
             cluster._maintenanceLoop = cluster.MaintenanceLoop();
+            cluster._logger.LogInformation("Raft node {Self} started; forwarding RPC on port +{Offset}.", options.Self, options.RpcPortOffset);
             return cluster;
         }
         catch
@@ -219,6 +231,7 @@ private async Task SeedConfigurationAsync(CancellationToken token)
 
         foreach (var member in _clusterMembers) config = config.Add(member);
         await _configStorage.SaveConfigurationAsync(config, configurationVersion: 0L, token).ConfigureAwait(false);
+        _logger.LogInformation("Seeded genesis cluster membership with {Members} member(s).", _clusterMembers.Count);
     }
 
     /// <summary>
@@ -235,6 +248,18 @@ private async Task SeedConfigurationAsync(CancellationToken token)
         // than a raw ObjectDisposedException from the cancellation source it is about to link.
         if (_lifetime.IsCancellationRequested) throw new RaftStorageException("The storage is shutting down.");
 
+        // A faulted state machine can no longer apply committed entries, so the local apply that completes
+        // this write's waiter will never run: the write would otherwise block for the full SubmitTimeout and
+        // then surface a misleading "ambiguous timeout" with no hint at the cause. Fail fast so the persistent
+        // fault is visible on every attempt (the root cause is logged once when the fault first happens) and
+        // points at the remedy. Recover the node by clearing its WAL to re-sync from the leader, or let an
+        // orchestrator restart it via the GetHealth().Faulted probe.
+        if (_stateMachine.IsFaulted)
+        {
+            _logger.LogWarning("Rejecting command {CommandId}: the local state machine has faulted and can no longer apply committed entries; recover this node (clear its WAL to re-sync from the leader).", command.Id);
+            throw new RaftStorageException("The local state machine has faulted; this node cannot serve writes until it is recovered.");
+        }
+
         var cluster = _cluster!; // non-null once StartAsync has returned, which is before any submit
 
         using var timeout = CancellationTokenSource.CreateLinkedTokenSource(token, _lifetime.Token);
@@ -262,13 +287,22 @@ private async Task SeedConfigurationAsync(CancellationToken token)
                         continue;
                     }
 
+                    // From here the command is being handed to the cluster and may be appended to the log, so
+                    // mark it appended NOW, before the await. This is what makes a timeout that fires DURING
+                    // the hand-off (the SubmitTimeout cancelling `linked` mid-replicate or mid-forward, which
+                    // throws an OperationCanceledException the inner catches deliberately do not convert) be
+                    // surfaced as ambiguous rather than as the safe-to-retry "no leader" outcome that would
+                    // let Hangfire retry a possibly-committed non-idempotent op. A provably-not-appended
+                    // failure (NotLeader / could-not-deliver / leader-wait timeout) is retryable and resets
+                    // this flag below before looping, so a never-appended command is not mislabelled ambiguous.
+                    appended = true;
+
                     if (leader.IsRemote)
                     {
                         // Forward by the leader's own endpoint form (a DnsEndPoint re-resolves, so a
                         // rescheduled leader is reached at its new IP without a restart).
                         var rpcEndpoint = RaftStorageOptions.RpcEndpoint(leader.EndPoint, _options.RpcPortOffset);
                         await _forwardingClient.SubmitAsync(rpcEndpoint, payload, linked).ConfigureAwait(false);
-                        appended = true;
                         break; // acknowledged as committed
                     }
 
@@ -289,17 +323,18 @@ private async Task SeedConfigurationAsync(CancellationToken token)
                         _logger.LogWarning(ex, "Local replication of command {CommandId} threw after append; treating the outcome as ambiguous and waiting for the local apply.", command.Id);
                     }
 
-                    appended = true;
                     break;
                 }
                 catch (AmbiguousCommandException ex)
                 {
                     _logger.LogWarning(ex, "Command {CommandId} outcome is ambiguous; waiting for the local apply to decide.", command.Id);
-                    appended = true;
-                    break;
+                    break; // appended is already true (set before the hand-off)
                 }
                 catch (Exception ex) when (IsRetryable(ex) && !linked.IsCancellationRequested)
                 {
+                    // Provably not appended (NotLeader / could-not-deliver / leader-wait timeout): safe to
+                    // resend, so re-classify the outcome as not-yet-handed-off before retrying.
+                    appended = false;
                     _logger.LogWarning(ex, "Command {CommandId} submission failed before reaching the leader, retrying.", command.Id);
                     await Task.Delay(TimeSpan.FromMilliseconds(100), linked).ConfigureAwait(false);
                 }
@@ -415,9 +450,17 @@ private async Task MaintenanceLoop()
             try
             {
                 await Task.Delay(_options.MaintenanceInterval, _lifetime.Token).ConfigureAwait(false);
+
+                // Every node samples its read staleness each interval, not just the leader: a partitioned
+                // follower keeps reporting HasLeader while its local apply falls behind the committed log and
+                // serves increasingly stale reads. This is the only place that hazard surfaces from logs
+                // alone; otherwise only an external GetHealth() poller can observe it.
+                WarnIfReadsAreStale();
+
                 if (_cluster?.Leader is { IsRemote: false })
                 {
-                    await SubmitAsync(Command.Single(new MaintenanceOp(_options.FetchInvisibilityTimeout)), _lifetime.Token).ConfigureAwait(false);
+                    var result = await SubmitAsync(Command.Single(new MaintenanceOp(_options.FetchInvisibilityTimeout)), _lifetime.Token).ConfigureAwait(false);
+                    if (result is MaintenanceSummary summary) LogMaintenance(summary);
                 }
             }
             catch (OperationCanceledException)
@@ -438,6 +481,44 @@ private async Task MaintenanceLoop()
         }
     }
 
+    /// <summary>
+    /// Surfaces what a maintenance pass changed. Runs only on the leader (the sole node that submits
+    /// maintenance), so each event is logged and metered once per pass cluster-wide rather than once per
+    /// replica. Stale-fetch reclaims are metered here authoritatively (the per-worker path in
+    /// <see cref="RaftFetchedJob"/> only sees reclaims of jobs whose worker is still alive; a crashed or
+    /// partitioned worker -- the dominant cause -- has no live renewer, so the leader is the only observer).
+    /// </summary>
+    private void LogMaintenance(MaintenanceSummary summary)
+    {
+        if (summary.StaleFetchesReclaimed > 0)
+        {
+            RaftMetrics.FetchLeaseReclaims.Add(summary.StaleFetchesReclaimed);
+            _logger.LogWarning("Maintenance reclaimed {Count} stale fetch lease(s) past the invisibility timeout; the affected job(s) will run again.", summary.StaleFetchesReclaimed);
+        }
+
+        // A dropped queue entry means an enqueued job was evicted (its expiry elapsed while it sat queued).
+        // That should not normally happen to an active job, so it is worth an operator's attention.
+        if (summary.OrphanedQueueEntriesRemoved > 0)
+            _logger.LogWarning("Maintenance dropped {Count} queue entry/entries pointing at an evicted (expired) job.", summary.OrphanedQueueEntriesRemoved);
+
+        if (summary.EvictedJobs > 0 || summary.ExpiredCollections > 0 || summary.ExpiredLocksReleased > 0)
+            _logger.LogDebug("Maintenance evicted {Jobs} expired job(s) and {Collections} expired collection(s), and released {Locks} expired lock(s).", summary.EvictedJobs, summary.ExpiredCollections, summary.ExpiredLocksReleased);
+    }
+
+    /// <summary>Warns when this node's local apply trails the committed log by more than the configured threshold, meaning its reads are stale.</summary>
+    private void WarnIfReadsAreStale()
+    {
+        var threshold = _options.ReadStalenessWarningThreshold;
+        if (threshold <= 0) return; // disabled
+
+        var wal = _wal;
+        if (wal is null) return;
+
+        var lag = wal.LastCommittedEntryIndex - wal.LastAppliedIndex;
+        if (lag > threshold)
+            _logger.LogWarning("Local apply is lagging the committed log by {Lag} entries (applied={Applied}, committed={Committed}); reads served by this node are stale.", lag, wal.LastAppliedIndex, wal.LastCommittedEntryIndex);
+    }
+
     public async ValueTask DisposeAsync()
     {
         await _lifetime.CancelAsync().ConfigureAwait(false);

src/Hangfire.Raft/RaftFetchedJob.cs

@@ -75,9 +75,11 @@ private async Task RenewAsync()
                 // A renewal that simply lost the race with a clean Ack/Requeue also gets `false` here
                 // (the lease is already gone). Only warn about a genuine reclaim of an in-flight job,
                 // not about one that already completed, to avoid a misleading "may run twice" log.
+                // The reclaim is metered on the maintenance side (RaftStorageCluster.MaintenanceLoop), which
+                // owns the count authoritatively so the dead-worker case (no live renewer to observe it) is
+                // also captured; this per-worker warning is the local breadcrumb for THIS job's double run.
                 if (!_completed && !_disposed)
                 {
-                    RaftMetrics.FetchLeaseReclaims.Add(1);
                     _cluster.Logger.LogWarning("Fetch lease for job {JobId} expired and was reclaimed; the job may run a second time.", JobId);
                 }
             }
@@ -113,9 +115,12 @@ public void Dispose()
             {
                 Requeue();
             }
-            catch (Exception)
+            catch (Exception ex)
             {
-                // The invisibility timeout reclaims the job if the requeue cannot reach the cluster.
+                // The invisibility timeout reclaims the job if the requeue cannot reach the cluster, so this
+                // is not a loss; but the job then stays invisible for the full FetchInvisibilityTimeout
+                // instead of being requeued promptly, so warn to make that delayed re-run correlatable.
+                _cluster.Logger.LogWarning(ex, "Could not requeue job {JobId} on dispose; it becomes visible again only after the fetch invisibility timeout.", JobId);
             }
         }
 

src/Hangfire.Raft/RaftJobStorage.cs

@@ -23,6 +23,7 @@ private RaftJobStorage(RaftStorageCluster cluster)
     public static async Task<RaftJobStorage> StartAsync(RaftStorageOptions options, CancellationToken cancellationToken = default)
     {
         ArgumentNullException.ThrowIfNull(options);
+        options.Validate();
         var cluster = await RaftStorageCluster.StartAsync(options, cancellationToken).ConfigureAwait(false);
         return new RaftJobStorage(cluster);
     }

src/Hangfire.Raft/RaftStorageConnection.cs

@@ -3,6 +3,7 @@
 using Hangfire.Raft.Commands;
 using Hangfire.Server;
 using Hangfire.Storage;
+using Microsoft.Extensions.Logging;
 
 namespace Hangfire.Raft;
 
@@ -67,9 +68,22 @@ public override IFetchedJob FetchNextJob(string[] queues, CancellationToken canc
                 if (Cluster.HasLeader && Cluster.Store.HasQueuedJob(queues))
                 {
                     var token = Guid.NewGuid();
-                    var result = Cluster.Submit(Command.Single(new FetchOp(queues, token)), cancellationToken);
-                    if (result is FetchResult fetched)
-                        return new RaftFetchedJob(Cluster, token, fetched.JobId, fetched.Queue);
+                    try
+                    {
+                        var result = Cluster.Submit(Command.Single(new FetchOp(queues, token)), cancellationToken);
+                        if (result is FetchResult fetched)
+                            return new RaftFetchedJob(Cluster, token, fetched.JobId, fetched.Queue);
+                    }
+                    catch (RaftStorageException ex)
+                    {
+                        // The fetch outcome could not be confirmed (no leader by the time it ran, or an
+                        // ambiguous timeout). If it actually committed it dequeued a job into a lease this
+                        // worker never received, so that job is invisible until maintenance reclaims it after
+                        // the invisibility timeout. Log so the stall is correlatable, then fall through to the
+                        // wait and re-probe once the cluster is healthy again (rather than surfacing the
+                        // exception to the worker, matching how the loop already handles a missing leader).
+                        Cluster.Logger.LogWarning(ex, "A fetch from queue(s) [{Queues}] could not be confirmed; if it committed, the dequeued job becomes visible again only after the fetch invisibility timeout.", string.Join(", ", queues));
+                    }
                 }
             }
             finally
@@ -123,6 +137,11 @@ public override void SetJobParameter(string id, string name, string value)
         }
         catch (JobLoadException ex)
         {
+            // A job whose invocation data no longer deserializes (the job type was removed/renamed, or there
+            // is a version skew between the enqueuing and executing deployments) can never run. The exception
+            // is surfaced to Hangfire via LoadException, but warn as well so the condition shows up in the
+            // operator's logs and not only when someone opens that specific job in the dashboard.
+            Cluster.Logger.LogWarning(ex, "Could not deserialize the invocation data for job {JobId}; its type may be missing or from an incompatible version, so it cannot be executed.", jobId);
             data.LoadException = ex;
         }