[windows] Safe worker shutdown signal handling

.gitignore

@@ -35,3 +35,5 @@ packages/cloud/
 /test_screenshots/
 third_party/
 packages/stem/workflow.sqlite*
+# Local serinus module (gitignored per user)
+packages/stem_serinus/

packages/dashboard/lib/src/server.dart

@@ -1031,34 +1031,36 @@ List<DashboardTaskStatusEntry> _applyTaskViewFilters(
   final namespaceFilter = options.namespaceFilter?.toLowerCase();
   final taskFilter = options.taskFilter?.toLowerCase();
   final runFilter = options.runId?.toLowerCase();
-  return tasks.where((entry) {
-    if (options.hasFilter) {
-      final queue = entry.queue.toLowerCase();
-      if (!(queueFilter != null && queue.contains(queueFilter))) {
-        return false;
-      }
-    }
-    if (options.hasNamespaceFilter &&
-        entry.namespace.toLowerCase() != namespaceFilter) {
-      return false;
-    }
-    if (options.hasTaskFilter) {
-      final name = entry.taskName.toLowerCase();
-      if (!(taskFilter != null && name.contains(taskFilter))) {
-        return false;
-      }
-    }
-    if (options.hasRunIdFilter) {
-      final runId = entry.runId?.toLowerCase() ?? '';
-      if (!(runFilter != null && runId.contains(runFilter))) {
-        return false;
-      }
-    }
-    if (options.hasStateFilter && entry.state != options.stateFilter) {
-      return false;
-    }
-    return true;
-  }).toList(growable: false);
+  return tasks
+      .where((entry) {
+        if (options.hasFilter) {
+          final queue = entry.queue.toLowerCase();
+          if (!(queueFilter != null && queue.contains(queueFilter))) {
+            return false;
+          }
+        }
+        if (options.hasNamespaceFilter &&
+            entry.namespace.toLowerCase() != namespaceFilter) {
+          return false;
+        }
+        if (options.hasTaskFilter) {
+          final name = entry.taskName.toLowerCase();
+          if (!(taskFilter != null && name.contains(taskFilter))) {
+            return false;
+          }
+        }
+        if (options.hasRunIdFilter) {
+          final runId = entry.runId?.toLowerCase() ?? '';
+          if (!(runFilter != null && runId.contains(runFilter))) {
+            return false;
+          }
+        }
+        if (options.hasStateFilter && entry.state != options.stateFilter) {
+          return false;
+        }
+        return true;
+      })
+      .toList(growable: false);
 }
 
 String _resolveDefaultNamespace(

packages/dashboard/lib/src/services/stem_service.dart

@@ -38,8 +38,9 @@ abstract class DashboardDataSource {
   Future<DashboardWorkflowRunSnapshot?> fetchWorkflowRun(String runId);
 
   /// Fetches persisted workflow checkpoints, if a workflow store is available.
-  Future<List<DashboardWorkflowCheckpointSnapshot>>
-  fetchWorkflowCheckpoints(String runId);
+  Future<List<DashboardWorkflowCheckpointSnapshot>> fetchWorkflowCheckpoints(
+    String runId,
+  );
 
   /// Enqueues a task request through the backing broker.
   Future<void> enqueueTask(EnqueueRequest request);

packages/dashboard/lib/src/ui/failures.dart

@@ -108,8 +108,7 @@ ${renderFailuresAlert(options)}
     options: DashboardTaskTableOptions(
       showState: false,
       emptyMessage: 'No individual failures to inspect.',
-      actionsBuilder: (task) =>
-          buildTaskReplayAction(task, redirectPath: redirectPath),
+      actionsBuilder: (task) => buildTaskReplayAction(task, redirectPath: redirectPath),
     ),
   )}
 </section>

packages/dashboard/lib/src/ui/jobs.dart

@@ -13,17 +13,19 @@ String buildJobsContent({
   final jobs = buildJobSummaries(taskStatuses, limit: 500);
   final taskFilter = options.task?.toLowerCase();
   final queueFilter = options.queue?.toLowerCase();
-  final filtered = jobs.where((entry) {
-    final matchesTask =
-        taskFilter == null ||
-        taskFilter.isEmpty ||
-        entry.taskName.toLowerCase().contains(taskFilter);
-    final matchesQueue =
-        queueFilter == null ||
-        queueFilter.isEmpty ||
-        entry.sampleQueue.toLowerCase().contains(queueFilter);
-    return matchesTask && matchesQueue;
-  }).toList(growable: false);
+  final filtered = jobs
+      .where((entry) {
+        final matchesTask =
+            taskFilter == null ||
+            taskFilter.isEmpty ||
+            entry.taskName.toLowerCase().contains(taskFilter);
+        final matchesQueue =
+            queueFilter == null ||
+            queueFilter.isEmpty ||
+            entry.sampleQueue.toLowerCase().contains(queueFilter);
+        return matchesTask && matchesQueue;
+      })
+      .toList(growable: false);
 
   final total = filtered.fold<int>(0, (sum, entry) => sum + entry.total);
   final running = filtered.fold<int>(0, (sum, entry) => sum + entry.running);

packages/dashboard/lib/src/ui/overview.dart

@@ -68,11 +68,9 @@ OverviewSections buildOverviewSections(
   List<QueueSummary> queues,
   List<WorkerStatus> workers,
   DashboardThroughput? throughput,
-  List<DashboardTaskStatusEntry> taskStatuses,
-  {
-    String defaultNamespace = 'stem',
-  }
-) {
+  List<DashboardTaskStatusEntry> taskStatuses, {
+  String defaultNamespace = 'stem',
+}) {
   final totalPending = queues.fold<int>(
     0,
     (total, summary) => total + summary.pending,
@@ -331,14 +329,14 @@ OverviewSections buildOverviewSections(
       '''
 <section id="overview-recent-tasks" class="table-card mt-7 ring-1 ring-inset ring-sky-300/10">
   ${buildTaskStatusTable(
-    taskStatuses.take(8).toList(growable: false),
-    options: const DashboardTaskTableOptions(
-      showAttempt: false,
-      showError: false,
-      showActions: false,
-      emptyMessage: 'No persisted task statuses yet.',
-    ),
-  )}
+        taskStatuses.take(8).toList(growable: false),
+        options: const DashboardTaskTableOptions(
+          showAttempt: false,
+          showError: false,
+          showActions: false,
+          emptyMessage: 'No persisted task statuses yet.',
+        ),
+      )}
 </section>
 ''';
 

packages/dashboard/lib/src/ui/tasks.dart

@@ -171,8 +171,7 @@ ${renderTasksAlert(options)}
     options: DashboardTaskTableOptions(
       emptyMessage: 'No task statuses match the current filters.',
       expandableRows: true,
-      actionsBuilder: (task) =>
-          buildTaskLifecycleActions(task, redirectPath: taskActionRedirect),
+      actionsBuilder: (task) => buildTaskLifecycleActions(task, redirectPath: taskActionRedirect),
     ),
   )}
   ${buildTaskPaginationBar(options, taskStatuses.length)}

packages/dashboard/pubspec.yaml

@@ -12,7 +12,7 @@ dependencies:
   ormed: ^0.2.0
   routed: ^0.3.2
   routed_hotwire: ^0.1.2
-  stem: ^0.2.0
+  stem: ">=0.2.0-dev <0.3.0"
   stem_cli: ^0.1.0
   stem_postgres: ^0.1.0
   stem_redis: ^0.1.0

packages/stem/CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 0.2.1-wip
+
+- Guarded worker and example process signal registration so Windows only
+  installs supported shutdown watchers, avoiding unsupported `SIGTERM` /
+  `SIGQUIT` subscriptions during graceful shutdown setup.
+
 ## 0.2.0
 
 - Added `StemClient.fromStack(...)` and `StemStack.createClient(...)` so

packages/stem/example/autoscaling_demo/bin/worker.dart

@@ -67,7 +67,9 @@ Future<void> main() async {
   }
 
   ProcessSignal.sigint.watch().listen(shutdown);
-  ProcessSignal.sigterm.watch().listen(shutdown);
+  if (!Platform.isWindows) {
+    ProcessSignal.sigterm.watch().listen(shutdown);
+  }
 
   await Completer<void>().future;
 }

packages/stem/example/daemonized_worker/bin/worker.dart

@@ -18,8 +18,10 @@ Future<void> main(List<String> args) async {
     }
   }
 
-  ProcessSignal.sigterm.watch().listen((_) => complete());
   ProcessSignal.sigint.watch().listen((_) => complete());
+  if (!Platform.isWindows) {
+    ProcessSignal.sigterm.watch().listen((_) => complete());
+  }
   // #endregion daemonized-worker-signal-handlers
 
   // Simulate background work.

packages/stem/example/dlq_sandbox/bin/worker.dart

@@ -36,7 +36,9 @@ Future<void> main() async {
   }
 
   ProcessSignal.sigint.watch().listen(shutdown);
-  ProcessSignal.sigterm.watch().listen(shutdown);
+  if (!Platform.isWindows) {
+    ProcessSignal.sigterm.watch().listen(shutdown);
+  }
 
   await worker.start();
 }

packages/stem/example/docs_snippets/lib/daemonization.dart

@@ -20,8 +20,10 @@ Future<void> main(List<String> args) async {
     }
   }
 
-  ProcessSignal.sigterm.watch().listen((_) => shutdown());
   ProcessSignal.sigint.watch().listen((_) => shutdown());
+  if (!Platform.isWindows) {
+    ProcessSignal.sigterm.watch().listen((_) => shutdown());
+  }
   // #endregion daemonization-signal-handlers
 
   // #region daemonization-loop

packages/stem/example/ecommerce/bin/server.dart

@@ -39,5 +39,7 @@ Future<void> main() async {
   }
 
   ProcessSignal.sigint.watch().listen(shutdown);
-  ProcessSignal.sigterm.watch().listen(shutdown);
+  if (!Platform.isWindows) {
+    ProcessSignal.sigterm.watch().listen(shutdown);
+  }
 }

packages/stem/example/email_service/bin/enqueuer.dart

@@ -25,7 +25,7 @@ Future<void> main(List<String> args) async {
       name: 'email.send',
       entrypoint: _placeholderEntrypoint,
       options: const TaskOptions(queue: 'emails', maxRetries: 3),
-      ),
+    ),
   ];
 
   final client = await StemClient.fromUrl(
@@ -82,7 +82,9 @@ Future<void> main(List<String> args) async {
   }
 
   ProcessSignal.sigint.watch().listen(shutdown);
-  ProcessSignal.sigterm.watch().listen(shutdown);
+  if (!Platform.isWindows) {
+    ProcessSignal.sigterm.watch().listen(shutdown);
+  }
 }
 
 FutureOr<Object?> _placeholderEntrypoint(

packages/stem/example/email_service/bin/worker.dart

@@ -71,7 +71,9 @@ Future<void> main(List<String> args) async {
   }
 
   ProcessSignal.sigint.watch().listen(shutdown);
-  ProcessSignal.sigterm.watch().listen(shutdown);
+  if (!Platform.isWindows) {
+    ProcessSignal.sigterm.watch().listen(shutdown);
+  }
 }
 
 Future<String> sendEmail(

packages/stem/example/encrypted_payload/worker/bin/worker.dart

@@ -58,7 +58,9 @@ Future<void> main(List<String> args) async {
   }
 
   ProcessSignal.sigint.watch().listen(shutdown);
-  ProcessSignal.sigterm.watch().listen(shutdown);
+  if (!Platform.isWindows) {
+    ProcessSignal.sigterm.watch().listen(shutdown);
+  }
 
   await Completer<void>().future;
 }

packages/stem/example/image_processor/bin/api.dart

@@ -73,7 +73,9 @@ Future<void> main(List<String> args) async {
   }
 
   ProcessSignal.sigint.watch().listen(shutdown);
-  ProcessSignal.sigterm.watch().listen(shutdown);
+  if (!Platform.isWindows) {
+    ProcessSignal.sigterm.watch().listen(shutdown);
+  }
 }
 
 FutureOr<Object?> _placeholderEntrypoint(

packages/stem/example/image_processor/bin/worker.dart

@@ -56,7 +56,9 @@ Future<void> main(List<String> args) async {
   }
 
   ProcessSignal.sigint.watch().listen(shutdown);
-  ProcessSignal.sigterm.watch().listen(shutdown);
+  if (!Platform.isWindows) {
+    ProcessSignal.sigterm.watch().listen(shutdown);
+  }
 }
 
 Future<String> generateThumbnail(

packages/stem/example/microservice/beat/bin/beat.dart

@@ -59,7 +59,9 @@ Future<void> main(List<String> args) async {
   }
 
   ProcessSignal.sigint.watch().listen(shutdown);
-  ProcessSignal.sigterm.watch().listen(shutdown);
+  if (!Platform.isWindows) {
+    ProcessSignal.sigterm.watch().listen(shutdown);
+  }
 
   await Completer<void>().future;
 }

packages/stem/example/microservice/enqueuer/bin/main.dart

@@ -260,7 +260,9 @@ Future<void> main(List<String> args) async {
   }
 
   ProcessSignal.sigint.watch().listen(shutdown);
-  ProcessSignal.sigterm.watch().listen(shutdown);
+  if (!Platform.isWindows) {
+    ProcessSignal.sigterm.watch().listen(shutdown);
+  }
 }
 
 FutureOr<Object?> _placeholderEntrypoint(

packages/stem/example/mixed_cluster/postgres_worker/bin/worker.dart

@@ -57,7 +57,9 @@ Future<void> main(List<String> args) async {
   }
 
   ProcessSignal.sigint.watch().listen(shutdown);
-  ProcessSignal.sigterm.watch().listen(shutdown);
+  if (!Platform.isWindows) {
+    ProcessSignal.sigterm.watch().listen(shutdown);
+  }
 
   await Completer<void>().future;
 }

packages/stem/example/mixed_cluster/redis_worker/bin/worker.dart

@@ -54,7 +54,9 @@ Future<void> main(List<String> args) async {
   }
 
   ProcessSignal.sigint.watch().listen(shutdown);
-  ProcessSignal.sigterm.watch().listen(shutdown);
+  if (!Platform.isWindows) {
+    ProcessSignal.sigterm.watch().listen(shutdown);
+  }
 
   await Completer<void>().future;
 }