diff --git a/neat_periodic_task/CHANGELOG.md b/neat_periodic_task/CHANGELOG.md index bdf1bf3c..7b573e70 100644 --- a/neat_periodic_task/CHANGELOG.md +++ b/neat_periodic_task/CHANGELOG.md @@ -1,3 +1,12 @@ +## v2.1.0 + * Added heartbeat support to `NeatPeriodicTaskScheduler`: + * `heartbeatInterval` and `heartbeatTimeout` parameters allow tasks to + periodically refresh a heartbeat timestamp in the status while running. + * If a running worker process crashes or is terminated abruptly, other + schedulers can detect the expired heartbeat after `heartbeatTimeout` + and reclaim the task without waiting for the full `timeout`. + * Added `heartbeat` property to `NeatTaskStatus`. + ## v2.0.1 * Added `topics` to `pubspec.yaml`. diff --git a/neat_periodic_task/lib/neat_periodic_task.dart b/neat_periodic_task/lib/neat_periodic_task.dart index 784f08b8..19c4df4b 100644 --- a/neat_periodic_task/lib/neat_periodic_task.dart +++ b/neat_periodic_task/lib/neat_periodic_task.dart @@ -14,7 +14,8 @@ library; -import 'dart:async' show Future, Completer, scheduleMicrotask, TimeoutException; +import 'dart:async' + show Future, Completer, scheduleMicrotask, TimeoutException, Timer; import 'package:logging/logging.dart' show Logger; import 'package:retry/retry.dart' show RetryOptions; @@ -114,6 +115,8 @@ class NeatPeriodicTaskScheduler { final NeatStatusProvider _statusProvider; final Duration _minCycle; final Duration _maxCycle; + final Duration? _heartbeatInterval; + final Duration? _heartbeatTimeout; bool _started = false; final _stopping = Completer(); @@ -145,6 +148,17 @@ class NeatPeriodicTaskScheduler { /// If the task fails consistently, it will be retried at [timeout] delay, /// this will continue indefinitely. Thus, it is sensible to pick a high /// [timeout], if the operation is expensive and this can be tolerated. + /// + /// If [heartbeatInterval] and/or [heartbeatTimeout] is provided, the scheduler + /// will periodically update a heartbeat timestamp in the status while the + /// [task] is running. Other schedulers will then detect if a running task has + /// stopped heartbeating for longer than [heartbeatTimeout] (e.g. because the + /// process crashed or was killed) and will reclaim and rerun the task + /// without having to wait for the full [timeout]. + /// + /// If only [heartbeatInterval] is provided, [heartbeatTimeout] defaults to + /// `3 * heartbeatInterval`. If only [heartbeatTimeout] is provided, + /// [heartbeatInterval] defaults to `heartbeatTimeout ~/ 3`. NeatPeriodicTaskScheduler({ required String name, required Duration interval, @@ -153,13 +167,19 @@ class NeatPeriodicTaskScheduler { NeatStatusProvider? status, Duration minCycle = const Duration(minutes: 5), Duration maxCycle = const Duration(hours: 3), + Duration? heartbeatInterval, + Duration? heartbeatTimeout, }) : _name = name, _interval = interval, _timeout = timeout, _task = task, _statusProvider = status ?? _InMemoryNeatStatusProvider(), _minCycle = minCycle, - _maxCycle = maxCycle { + _maxCycle = maxCycle, + _heartbeatInterval = heartbeatInterval ?? + (heartbeatTimeout != null ? heartbeatTimeout ~/ 3 : null), + _heartbeatTimeout = heartbeatTimeout ?? + (heartbeatInterval != null ? heartbeatInterval * 3 : null) { if (maxCycle <= minCycle) { throw ArgumentError.value( maxCycle, 'maxCycle', 'maxCycle must larger than minCycle'); @@ -168,6 +188,14 @@ class NeatPeriodicTaskScheduler { throw ArgumentError.value(interval, 'interval', 'interval must be large than 2 * minCycle for reasonable behavior'); } + if (_heartbeatInterval != null && _heartbeatInterval <= Duration.zero) { + throw ArgumentError.value(_heartbeatInterval, 'heartbeatInterval', + 'heartbeatInterval must be positive'); + } + if (_heartbeatTimeout != null && _heartbeatTimeout <= _heartbeatInterval!) { + throw ArgumentError.value(_heartbeatTimeout, 'heartbeatTimeout', + 'heartbeatTimeout must be larger than heartbeatInterval'); + } } /// Start the scheduler. @@ -239,48 +267,75 @@ class NeatPeriodicTaskScheduler { return; } - // Find time elapsed since last time the task started running. final now = DateTime.now().toUtc(); - final elapsed = now.difference(status.started); - _log.finest(() => 'time elapsed since "$_name" was last started $elapsed'); - // If state is 'finished' the delay before next run is _interval, otherwise - // we assume state is 'running' as delay is only _timeout. - var delay = _interval; - if (status.state != 'finished') { - delay = _timeout; - } + if (status.state == 'finished') { + // Find time elapsed since last time the task started running. + final elapsed = now.difference(status.started); + _log.finest( + () => 'time elapsed since "$_name" was last started $elapsed'); + + // If state is 'finished' the delay before next run is _interval. + final delay = _interval; + if (elapsed < delay) { + var d = delay ~/ 2; + // Always sleep at least minCycle to ensure the iteration doesn't spin too + // fast as we approach the next iteration. + if (d < _minCycle) { + d = _minCycle; + } + // Never sleep more than maxCycle, as we must wake-up and print a log line + // that says we've checked the status of the task. Operators can either + // monitor the message saying this was done, or they can monitor the + // message saying that the status was monitored. + if (d > _maxCycle) { + d = _maxCycle; + } + _log.info('### [ALIVE] neat-periodic-task: "$_name"'); + _log.finest(() => 'NeatPeriodicTaskScheduler "$_name" sleeps $d'); + await _sleep(d); - // If delay isn't past yet, we sleep. - if (elapsed < delay) { - var d = delay ~/ 2; - // Always sleep at least minCycle to ensure the iteration doesn't spin too - // fast as we approach the next iteration. - if (d < _minCycle) { - d = _minCycle; + // Return such that we do another _iteration() call.. + return; } - // Never sleep more than maxCycle, as we must wake-up and print a log line - // that says we've checked the status of the task. Operators can either - // monitor the message saying this was done, or they can monitor the - // message saying that the status was monitored. - if (d > _maxCycle) { - d = _maxCycle; + } else if (status.state == 'running') { + final totalElapsed = now.difference(status.started); + final isTimeoutExpired = totalElapsed >= _timeout; + final heartbeat = status.heartbeat; + final isHeartbeatExpired = _heartbeatTimeout != null && + heartbeat != null && + now.difference(heartbeat) >= _heartbeatTimeout; + + if (!isTimeoutExpired && !isHeartbeatExpired) { + var remaining = _timeout - totalElapsed; + if (_heartbeatTimeout != null && heartbeat != null) { + final heartbeatRemaining = + _heartbeatTimeout - now.difference(heartbeat); + if (heartbeatRemaining < remaining) { + remaining = heartbeatRemaining; + } + } + var d = remaining ~/ 2; + if (d < _minCycle) { + d = _minCycle; + } + if (d > _maxCycle) { + d = _maxCycle; + } + _log.finest( + () => 'NeatPeriodicTaskScheduler "$_name" is running, sleeps $d'); + await _sleep(d); + + return; } - if (status.state == 'finished') { - _log.info('### [ALIVE] neat-periodic-task: "$_name"'); - } - _log.finest(() => 'NeatPeriodicTaskScheduler "$_name" sleeps $d'); - await _sleep(d); - - // Return such that we do another _iteration() call.. - return; } - // If elapsed >= delay, then we claim and run the task. + // If state is not running (or lock/heartbeat expired), we claim and run. await _claimAndRun(status.update( owner: Slugid.nice().toString(), state: 'running', started: now, + heartbeat: now, )); } @@ -294,6 +349,37 @@ class NeatPeriodicTaskScheduler { return; } + var currentStatus = status; + Timer? heartbeatTimer; + Future? pendingHeartbeat; + var isSendingHeartbeat = false; + var isRunning = true; + + if (_heartbeatInterval != null) { + heartbeatTimer = Timer.periodic(_heartbeatInterval, (_) { + if (!isRunning || isSendingHeartbeat) return; + isSendingHeartbeat = true; + pendingHeartbeat = () async { + try { + final now = DateTime.now().toUtc(); + _log.finest(() => 'Sending heartbeat for "$_name"'); + final nextStatus = currentStatus.update(heartbeat: now); + final ok = await _statusProvider.set(nextStatus.serialize()); + if (ok) { + currentStatus = nextStatus; + } else { + _log.warning( + 'Failed to send heartbeat for "$_name", lock may be lost'); + } + } catch (e, st) { + _log.warning('Error sending heartbeat for "$_name"', e, st); + } finally { + isSendingHeartbeat = false; + } + }(); + }); + } + try { _log.info('### [START] neat-periodic-task: "$_name"'); await _task().timeout(_timeout); @@ -305,10 +391,14 @@ class NeatPeriodicTaskScheduler { } catch (e, st) { _log.shout('### [FAILED] neat-periodic-task: "$_name"', e, st); return; + } finally { + isRunning = false; + heartbeatTimer?.cancel(); + await pendingHeartbeat; } _log.finest(() => 'Attempting to set finished status for "$_name"'); - final st = status.update(state: 'finished').serialize(); + final st = currentStatus.update(state: 'finished').serialize(); if (!await _statusProvider.set(st)) { _log.warning( 'Failed to set finished status for "$_name" ' @@ -330,13 +420,19 @@ class NeatPeriodicTaskScheduler { // Find time elapsed since last time the task was started. final now = DateTime.now().toUtc(); final elapsed = now.difference(status.started); + final isTimeoutExpired = elapsed > _timeout; + final heartbeat = status.heartbeat; + final isHeartbeatExpired = _heartbeatTimeout != null && + heartbeat != null && + now.difference(heartbeat) > _heartbeatTimeout; // If not running, or timed-out we run the task again. - if (status.state != 'running' || elapsed > _timeout) { + if (status.state != 'running' || isTimeoutExpired || isHeartbeatExpired) { await _claimAndRun(status.update( owner: Slugid.nice().toString(), state: 'running', started: now, + heartbeat: now, )); } else { _log.info('trigger() call on "$_name" ignored, as task is running'); diff --git a/neat_periodic_task/lib/src/neat_status.dart b/neat_periodic_task/lib/src/neat_status.dart index baa92a75..1466c5a7 100644 --- a/neat_periodic_task/lib/src/neat_status.dart +++ b/neat_periodic_task/lib/src/neat_status.dart @@ -46,6 +46,11 @@ class NeatTaskStatus { /// Time the task started running. final DateTime started; + /// Time of the last heartbeat sent by the running task. + /// + /// Only relevant if [state] is 'running'. + final DateTime? heartbeat; + /// Owner if this status object, only relevant if [state] is 'running'. /// /// The [owner] is a random slugid that identifies the process that owns the @@ -57,12 +62,14 @@ class NeatTaskStatus { required this.version, required this.state, required this.started, + this.heartbeat, required this.owner, }); NeatTaskStatus.create({ required this.state, required this.started, + this.heartbeat, required this.owner, }) : format = formatIdentifier, version = currentVersion; @@ -71,6 +78,7 @@ class NeatTaskStatus { NeatTaskStatus update({ String? state, DateTime? started, + DateTime? heartbeat, String? owner, }) { return NeatTaskStatus( @@ -78,6 +86,7 @@ class NeatTaskStatus { version: currentVersion, state: state ?? this.state, started: started ?? this.started, + heartbeat: heartbeat ?? this.heartbeat, owner: owner ?? this.owner, ); } @@ -87,6 +96,7 @@ class NeatTaskStatus { version: currentVersion, state: 'idle', started: DateTime.fromMillisecondsSinceEpoch(0, isUtc: true), + heartbeat: null, owner: '-', ); diff --git a/neat_periodic_task/lib/src/neat_status.g.dart b/neat_periodic_task/lib/src/neat_status.g.dart index ecb915ac..d922582c 100644 --- a/neat_periodic_task/lib/src/neat_status.g.dart +++ b/neat_periodic_task/lib/src/neat_status.g.dart @@ -12,6 +12,9 @@ NeatTaskStatus _$NeatTaskStatusFromJson(Map json) { version: json['version'] as int, state: json['state'] as String, started: DateTime.parse(json['started'] as String), + heartbeat: json['heartbeat'] == null + ? null + : DateTime.parse(json['heartbeat'] as String), owner: json['owner'] as String, ); } @@ -22,5 +25,7 @@ Map _$NeatTaskStatusToJson(NeatTaskStatus instance) => 'version': instance.version, 'state': instance.state, 'started': instance.started.toIso8601String(), + if (instance.heartbeat != null) + 'heartbeat': instance.heartbeat!.toIso8601String(), 'owner': instance.owner, }; diff --git a/neat_periodic_task/pubspec.yaml b/neat_periodic_task/pubspec.yaml index 6075304d..26a761f0 100644 --- a/neat_periodic_task/pubspec.yaml +++ b/neat_periodic_task/pubspec.yaml @@ -1,5 +1,5 @@ name: neat_periodic_task -version: 2.0.1 +version: 2.1.0 description: >- Auxiliary classes for reliably running a periodic task in a long-running process such as web-server. diff --git a/neat_periodic_task/test/neat_periodic_task_test.dart b/neat_periodic_task/test/neat_periodic_task_test.dart index 9070e51f..9f745bc1 100644 --- a/neat_periodic_task/test/neat_periodic_task_test.dart +++ b/neat_periodic_task/test/neat_periodic_task_test.dart @@ -12,9 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. -import 'dart:async' show Future; +import 'dart:async' show Future, Completer; +import 'dart:convert' show json, utf8; import 'package:test/test.dart'; import 'package:neat_periodic_task/neat_periodic_task.dart'; +import 'package:neat_periodic_task/src/neat_status.dart'; import 'package:logging/logging.dart'; import 'package:collection/collection.dart' show ListEquality; @@ -120,4 +122,421 @@ void main() { expect(count, inInclusiveRange(6, 7)); }); + + test('heartbeat updates while task is running', () async { + final statusStore = _StatusStore(); + final taskStarted = Completer(); + final allowFinish = Completer(); + + final scheduler = NeatPeriodicTaskScheduler( + name: 'heartbeat-test', + interval: Duration(milliseconds: 500), + timeout: Duration(seconds: 10), + minCycle: Duration(milliseconds: 50), + maxCycle: Duration(milliseconds: 100), + heartbeatInterval: Duration(milliseconds: 30), + heartbeatTimeout: Duration(milliseconds: 100), + status: statusStore.provider(), + task: () async { + taskStarted.complete(); + await allowFinish.future; + }, + ); + + scheduler.start(); + await taskStarted.future; + + // Give it time to send a few heartbeats. + await Future.delayed(Duration(milliseconds: 40)); + final status1 = NeatTaskStatus.deserialize(statusStore._value); + expect(status1.state, equals('running')); + expect(status1.heartbeat, isNotNull); + + await Future.delayed(Duration(milliseconds: 60)); + final status2 = NeatTaskStatus.deserialize(statusStore._value); + expect(status2.state, equals('running')); + expect(status2.heartbeat!.isAfter(status1.heartbeat!), isTrue); + + allowFinish.complete(); + await Future.delayed(Duration(milliseconds: 50)); + await scheduler.stop(); + + final statusFinal = NeatTaskStatus.deserialize(statusStore._value); + expect(statusFinal.state, equals('finished')); + }); + + test( + 'abandoned task with expired heartbeat is reclaimed without waiting for full timeout', + () async { + final statusStore = _StatusStore(); + final now = DateTime.now().toUtc(); + + // Simulate an abandoned task that started 1 minute ago, last heartbeat 200ms ago. + // Full timeout is 1 hour! + final abandoned = NeatTaskStatus.create( + state: 'running', + started: now.subtract(Duration(minutes: 1)), + heartbeat: now.subtract(Duration(milliseconds: 200)), + owner: 'dead-worker', + ); + statusStore._value = abandoned.serialize(); + + var reclaimed = false; + final scheduler = NeatPeriodicTaskScheduler( + name: 'recovery-test', + interval: Duration(seconds: 10), + timeout: Duration(hours: 1), + minCycle: Duration(milliseconds: 50), + maxCycle: Duration(milliseconds: 100), + heartbeatInterval: Duration(milliseconds: 30), + heartbeatTimeout: Duration(milliseconds: 100), + status: statusStore.provider(), + task: () async { + reclaimed = true; + }, + ); + + scheduler.start(); + await Future.delayed(Duration(milliseconds: 150)); + await scheduler.stop(); + + expect(reclaimed, isTrue); + final status = NeatTaskStatus.deserialize(statusStore._value); + expect(status.state, equals('finished')); + expect(status.owner, isNot(equals('dead-worker'))); + }); + + test('trigger reclaims task if heartbeat is expired', () async { + final statusStore = _StatusStore(); + final now = DateTime.now().toUtc(); + + // Abandoned task with expired heartbeat, but overall timeout not reached. + final abandoned = NeatTaskStatus.create( + state: 'running', + started: now.subtract(Duration(minutes: 1)), + heartbeat: now.subtract(Duration(milliseconds: 200)), + owner: 'dead-worker', + ); + statusStore._value = abandoned.serialize(); + + var ran = false; + final scheduler = NeatPeriodicTaskScheduler( + name: 'trigger-test', + interval: Duration(seconds: 10), + timeout: Duration(hours: 1), + minCycle: Duration(milliseconds: 50), + maxCycle: Duration(milliseconds: 100), + heartbeatInterval: Duration(milliseconds: 30), + heartbeatTimeout: Duration(milliseconds: 100), + status: statusStore.provider(), + task: () async { + ran = true; + }, + ); + + await scheduler.trigger(); + expect(ran, isTrue); + final status = NeatTaskStatus.deserialize(statusStore._value); + expect(status.state, equals('finished')); + }); + + test('active task with healthy heartbeats is not stolen', () async { + final statusStore = _StatusStore(); + final aStarted = Completer(); + final allowAFinish = Completer(); + var bRan = false; + + final schedulerA = NeatPeriodicTaskScheduler( + name: 'machine-A', + interval: Duration(seconds: 10), + timeout: Duration(seconds: 5), + minCycle: Duration(milliseconds: 50), + maxCycle: Duration(milliseconds: 100), + heartbeatInterval: Duration(milliseconds: 30), + heartbeatTimeout: Duration(milliseconds: 100), + status: statusStore.provider(), + task: () async { + aStarted.complete(); + await allowAFinish.future; + }, + ); + + final schedulerB = NeatPeriodicTaskScheduler( + name: 'machine-B', + interval: Duration(seconds: 10), + timeout: Duration(seconds: 5), + minCycle: Duration(milliseconds: 50), + maxCycle: Duration(milliseconds: 100), + heartbeatInterval: Duration(milliseconds: 30), + heartbeatTimeout: Duration(milliseconds: 100), + status: statusStore.provider(), + task: () async { + bRan = true; + }, + ); + + schedulerA.start(); + await aStarted.future; + + schedulerB.start(); + // Allow A to run for ~150ms while sending heartbeats every 30ms. + // Since heartbeatTimeout is 100ms and A sends heartbeats every 30ms, B must not run. + await Future.delayed(Duration(milliseconds: 150)); + expect(bRan, isFalse); + + allowAFinish.complete(); + await Future.delayed(Duration(milliseconds: 50)); + await schedulerA.stop(); + await schedulerB.stop(); + + expect(bRan, isFalse); + final status = NeatTaskStatus.deserialize(statusStore._value); + expect(status.state, equals('finished')); + }); + + test('overall timeout terminates task even if heartbeats are active', + () async { + final statusStore = _StatusStore(); + final taskStarted = Completer(); + + final scheduler = NeatPeriodicTaskScheduler( + name: 'timeout-override-test', + interval: Duration(seconds: 10), + timeout: Duration(milliseconds: 100), + minCycle: Duration(milliseconds: 50), + maxCycle: Duration(milliseconds: 100), + heartbeatInterval: Duration(milliseconds: 20), + heartbeatTimeout: Duration(milliseconds: 80), + status: statusStore.provider(), + task: () async { + taskStarted.complete(); + // Hang indefinitely while heartbeats fire + await Completer().future; + }, + ); + + scheduler.start(); + await taskStarted.future; + + // After > 100ms, timeout should trigger and terminate task + await Future.delayed(Duration(milliseconds: 200)); + await scheduler.stop(); + + // Since timeout occurred, the task failed and was not marked 'finished'. + final status = NeatTaskStatus.deserialize(statusStore._value); + expect(status.state, equals('running')); + }); + + test( + 'task with active heartbeats is reclaimed if total duration exceeds timeout', + () async { + final statusStore = _StatusStore(); + final now = DateTime.now().toUtc(); + + // Task started 200ms ago, but heartbeat is fresh (10ms ago). + // However, total timeout is only 100ms! + final stuckTask = NeatTaskStatus.create( + state: 'running', + started: now.subtract(Duration(milliseconds: 200)), + heartbeat: now.subtract(Duration(milliseconds: 10)), + owner: 'stuck-worker', + ); + statusStore._value = stuckTask.serialize(); + + var reclaimed = false; + final scheduler = NeatPeriodicTaskScheduler( + name: 'stuck-reclaim-test', + interval: Duration(seconds: 10), + timeout: Duration(milliseconds: 100), + minCycle: Duration(milliseconds: 50), + maxCycle: Duration(milliseconds: 100), + heartbeatInterval: Duration(milliseconds: 20), + heartbeatTimeout: Duration(milliseconds: 80), + status: statusStore.provider(), + task: () async { + reclaimed = true; + }, + ); + + scheduler.start(); + await Future.delayed(Duration(milliseconds: 100)); + await scheduler.stop(); + + expect(reclaimed, isTrue); + final status = NeatTaskStatus.deserialize(statusStore._value); + expect(status.state, equals('finished')); + expect(status.owner, isNot(equals('stuck-worker'))); + }); + + test( + 'status without heartbeat falls back to full timeout and is not prematurely reclaimed', + () async { + final statusStore = _StatusStore(); + final now = DateTime.now().toUtc(); + + // Legacy status: started 200ms ago, NO heartbeat. + // heartbeatTimeout is 100ms, but full timeout is 10 seconds! + final legacyStatus = NeatTaskStatus( + format: NeatTaskStatus.formatIdentifier, + version: NeatTaskStatus.currentVersion, + state: 'running', + started: now.subtract(Duration(milliseconds: 200)), + heartbeat: null, + owner: 'legacy-worker', + ); + statusStore._value = legacyStatus.serialize(); + + var reclaimed = false; + final scheduler = NeatPeriodicTaskScheduler( + name: 'legacy-fallback-test', + interval: Duration(seconds: 10), + timeout: Duration(seconds: 10), + minCycle: Duration(milliseconds: 50), + maxCycle: Duration(milliseconds: 100), + heartbeatInterval: Duration(milliseconds: 30), + heartbeatTimeout: Duration(milliseconds: 100), + status: statusStore.provider(), + task: () async { + reclaimed = true; + }, + ); + + scheduler.start(); + // Wait for longer than heartbeatTimeout (100ms) + await Future.delayed(Duration(milliseconds: 250)); + await scheduler.stop(); + + // Must NOT have reclaimed the task because legacy status lacks heartbeat! + expect(reclaimed, isFalse); + final status = NeatTaskStatus.deserialize(statusStore._value); + expect(status.owner, equals('legacy-worker')); + expect(status.state, equals('running')); + }); + + test('heartbeat timer is cancelled when task throws an exception', () async { + final statusStore = _StatusStore(); + + final scheduler = NeatPeriodicTaskScheduler( + name: 'exception-cleanup-test', + interval: Duration(seconds: 10), + timeout: Duration(seconds: 5), + minCycle: Duration(milliseconds: 50), + maxCycle: Duration(milliseconds: 100), + heartbeatInterval: Duration(milliseconds: 20), + heartbeatTimeout: Duration(seconds: 10), + status: statusStore.provider(), + task: () async { + await Future.delayed(Duration(milliseconds: 10)); + throw Exception('task failed'); + }, + ); + + scheduler.start(); + await Future.delayed(Duration(milliseconds: 50)); + final statusAfterFailure = NeatTaskStatus.deserialize(statusStore._value); + final lastHeartbeat = statusAfterFailure.heartbeat; + + // Wait more intervals; no more heartbeats should be written + await Future.delayed(Duration(milliseconds: 80)); + final statusLater = NeatTaskStatus.deserialize(statusStore._value); + expect(statusLater.heartbeat, equals(lastHeartbeat)); + + await scheduler.stop(); + }); + + test('loss of lock during heartbeat is handled gracefully without crashing', + () async { + final statusStore = _StatusStore(); + final taskStarted = Completer(); + final allowFinish = Completer(); + + final scheduler = NeatPeriodicTaskScheduler( + name: 'lock-loss-test', + interval: Duration(seconds: 10), + timeout: Duration(seconds: 5), + minCycle: Duration(milliseconds: 50), + maxCycle: Duration(milliseconds: 100), + heartbeatInterval: Duration(milliseconds: 20), + heartbeatTimeout: Duration(milliseconds: 80), + status: statusStore.provider(), + task: () async { + taskStarted.complete(); + await allowFinish.future; + }, + ); + + scheduler.start(); + await taskStarted.future; + + // Simulate another worker stealing the lock while task is running + final stolenStatus = NeatTaskStatus.create( + state: 'running', + started: DateTime.now().toUtc(), + owner: 'thief-worker', + ); + statusStore._value = stolenStatus.serialize(); + + // Allow heartbeats to try to send while lock is stolen + await Future.delayed(Duration(milliseconds: 60)); + + // Finish task and stop scheduler + allowFinish.complete(); + await Future.delayed(Duration(milliseconds: 30)); + await scheduler.stop(); + + // The thief's lock should NOT have been overwritten with finished + final currentStatus = NeatTaskStatus.deserialize(statusStore._value); + expect(currentStatus.owner, equals('thief-worker')); + }); + + group('NeatTaskStatus serialization', () { + test('round-trips with heartbeat', () { + final now = DateTime.now().toUtc(); + final status = NeatTaskStatus.create( + state: 'running', + started: now, + heartbeat: now.add(Duration(seconds: 1)), + owner: 'test-owner', + ); + final bytes = status.serialize(); + final decoded = NeatTaskStatus.deserialize(bytes); + expect(decoded.format, equals(NeatTaskStatus.formatIdentifier)); + expect(decoded.version, equals(NeatTaskStatus.currentVersion)); + expect(decoded.state, equals('running')); + expect(decoded.started, equals(now)); + expect(decoded.heartbeat, equals(now.add(Duration(seconds: 1)))); + expect(decoded.owner, equals('test-owner')); + }); + + test('deserializes legacy JSON without heartbeat field', () { + final jsonMap = { + 'format': NeatTaskStatus.formatIdentifier, + 'version': 1, + 'state': 'running', + 'started': '2026-08-28T10:00:00.000Z', + 'owner': 'legacy-owner', + }; + final bytes = json.fuse(utf8).encode(jsonMap); + final decoded = NeatTaskStatus.deserialize(bytes); + expect(decoded.format, equals(NeatTaskStatus.formatIdentifier)); + expect(decoded.state, equals('running')); + expect(decoded.heartbeat, isNull); + expect(decoded.owner, equals('legacy-owner')); + }); + + test('deserializes JSON with explicit null heartbeat', () { + final jsonMap = { + 'format': NeatTaskStatus.formatIdentifier, + 'version': 1, + 'state': 'finished', + 'started': '2026-08-28T10:00:00.000Z', + 'heartbeat': null, + 'owner': 'legacy-owner', + }; + final bytes = json.fuse(utf8).encode(jsonMap); + final decoded = NeatTaskStatus.deserialize(bytes); + expect(decoded.state, equals('finished')); + expect(decoded.heartbeat, isNull); + }); + }); }