diff --git a/ravendb/__init__.py b/ravendb/__init__.py index b4e6c79b..7c2def9a 100644 --- a/ravendb/__init__.py +++ b/ravendb/__init__.py @@ -850,6 +850,10 @@ SubscriptionBatch, SubscriptionWorker, ) +from ravendb.documents.subscriptions.worker_status import ( + SubscriptionWorkerState, + SubscriptionWorkerStatus, +) from ravendb.documents.time_series import TimeSeriesOperations # todo: Serverwide diff --git a/ravendb/documents/subscriptions/worker.py b/ravendb/documents/subscriptions/worker.py index 4d989643..5d0d5494 100644 --- a/ravendb/documents/subscriptions/worker.py +++ b/ravendb/documents/subscriptions/worker.py @@ -21,6 +21,7 @@ from ravendb.documents.session.document_info import DocumentInfo from ravendb.documents.session.misc import SessionOptions, TransactionMode from ravendb.documents.subscriptions.revision import Revision +from ravendb.documents.subscriptions.worker_status import SubscriptionWorkerState, SubscriptionWorkerStatus from ravendb.exceptions.cluster import NodeIsPassiveException from ravendb.exceptions.exceptions import ( AuthorizationException, @@ -186,6 +187,11 @@ def __init__( self._after_acknowledgment: List[Callable[[SubscriptionBatch[_T]], None]] = [] self._on_subscription_connection_retry: List[Callable[[Exception], None]] = [] self._on_unexpected_subscription_error: List[Callable[[Exception], None]] = [] + self._on_established_subscription_connection: List[Callable[[], None]] = [] + self._on_state_changed: List[Callable[[SubscriptionWorker[_T], SubscriptionWorkerStatus], None]] = [] + + self._failing_since_utc: Optional[datetime.datetime] = None + self._status = SubscriptionWorkerStatus(SubscriptionWorkerState.NOT_STARTED, None, self._utc_now(), None) self._redirect_node: Optional[ServerNode] = None self._subscription_local_request_executor: Optional[RequestExecutor] = None @@ -233,6 +239,83 @@ def invoke_on_unexpected_subscription_error(self, err: Exception) -> None: for event in self._on_unexpected_subscription_error: event(err) + def add_on_established_subscription_connection(self, handler: Callable[[], None]) -> None: + """ + Registers a handler raised whenever the worker has successfully connected to the server, both for the + initial connection and for every reconnection. Pair it with on_subscription_connection_retry to track + whether the worker is currently connected, instead of waiting for the next batch to arrive. + """ + self._on_established_subscription_connection.append(handler) + + def remove_on_established_subscription_connection(self, handler: Callable[[], None]) -> None: + self._on_established_subscription_connection.remove(handler) + + def invoke_on_established_subscription_connection(self) -> None: + for event in self._on_established_subscription_connection: + event() + + @property + def status(self) -> SubscriptionWorkerStatus: + """ + What the worker is doing right now: connecting, waiting for the server to send documents, processing a + batch, retrying after a failure, or stopped. The returned snapshot is immutable, so the state and the + exception that produced it always belong together. + """ + return self._status + + def add_on_state_changed(self, handler: Callable[[SubscriptionWorker[_T], SubscriptionWorkerStatus], None]) -> None: + """ + Registers a handler raised on every change to status. The worker that changed state is passed along with + the snapshot that was just installed, so that a single handler can serve several workers. Exceptions + raised by a handler are logged and swallowed, so that watching the worker cannot break it. + """ + self._on_state_changed.append(handler) + + def remove_on_state_changed( + self, handler: Callable[[SubscriptionWorker[_T], SubscriptionWorkerStatus], None] + ) -> None: + self._on_state_changed.remove(handler) + + @staticmethod + def _utc_now() -> datetime.datetime: + return datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) + + # Called only from the thread running the subscription (or from close() on a worker that was never run) + def _set_state(self, state: SubscriptionWorkerState, exception: Optional[Exception] = None) -> None: + now = self._utc_now() + + if state in (SubscriptionWorkerState.WAITING_FOR_DOCUMENTS, SubscriptionWorkerState.PROCESSING): + # talking to the server, whatever went wrong before is over + self._failing_since_utc = None + elif state in (SubscriptionWorkerState.RETRYING, SubscriptionWorkerState.FAULTED): + # keep the first failure time, so that it shows how long the worker has been failing + if self._failing_since_utc is None: + self._failing_since_utc = now + # CONNECTING is a step of every retry as well as of a healthy start, so it carries the failure streak over + + current = self._status + if ( + current.state == state + and current.exception is exception + and current.failing_since_utc == self._failing_since_utc + ): + return # same state, nothing to change + + status = SubscriptionWorkerStatus(state, exception, now, self._failing_since_utc) + self._status = status + + for handler in list(self._on_state_changed): + try: + handler(self, status) + except Exception as e: + # a failing handler must not take the subscription down with it, and must not replace the + # exception that is being reported + self._logger.info( + f"Subscription '{self._options.subscription_name}'. " + f"on_state_changed handler raised while reporting '{status}'.", + exc_info=e, + ) + def close(self, wait_for_subscription_task: bool = True) -> None: if self._disposed: return @@ -256,6 +339,11 @@ def close(self, wait_for_subscription_task: bool = True) -> None: self._logger.debug(f"Error during close of subscription: {ex.args[0]}", ex) finally: + if self._subscription_task is None: + # a worker that was never run has no task to report STOPPED for itself, + # and cannot be racing with one either + self._set_state(SubscriptionWorkerState.STOPPED) + if self._on_closed is not None: self._on_closed(self) @@ -536,6 +624,11 @@ def _process_subscription(self) -> None: if self._processing_cts.get_token().is_cancellation_requested(): return + # set before raising the event, so that a handler reading status sees that the worker is connected + self._set_state(SubscriptionWorkerState.WAITING_FOR_DOCUMENTS) + + self.invoke_on_established_subscription_connection() + notified_subscriber = concurrent.futures.Future() notified_subscriber.set_result(None) @@ -584,6 +677,12 @@ def __run_async(): raise e + # The read above was started before the subscriber finished, on purpose, so the state stays + # PROCESSING while it is in flight. Now that the subscriber is done, the server is the only thing + # left to wait for, unless its batch already arrived in the meantime. + if not read_from_server.done(): + self._set_state(SubscriptionWorkerState.WAITING_FOR_DOCUMENTS) + incoming_batch = read_from_server.result( self._options.time_to_wait_before_connection_retry.total_seconds() ) @@ -592,6 +691,8 @@ def __run_async(): last_received_change_vector = batch.initialize(incoming_batch) + self._set_state(SubscriptionWorkerState.PROCESSING) + notified_subscriber = self._store.thread_pool_executor.submit(__run_async) except OperationCancelledException as e: if not self._disposed: @@ -687,9 +788,24 @@ def _send_ack(self, last_received_change_vector: str, network_stream: socket) -> def _run_subscription_async(self) -> Future[None]: def __run_async() -> None: + try: + __run_internal() + except Exception as e: + # the worker was asked to stop, the failure is not a problem + if not self._disposed: + self._set_state(SubscriptionWorkerState.FAULTED, e) + raise + finally: + # reached on the paths that end the worker without surfacing an error (closing the worker), + # while a worker that gave up on a failure keeps reporting it + if self._status.state != SubscriptionWorkerState.FAULTED: + self._set_state(SubscriptionWorkerState.STOPPED) + + def __run_internal() -> None: while not self._processing_cts.get_token().is_cancellation_requested(): try: self._close_tcp_client() + self._set_state(SubscriptionWorkerState.CONNECTING) self._logger.info(f"Subscription {self._options.subscription_name}. Connection to server...") self._process_subscription() @@ -706,6 +822,9 @@ def __run_async() -> None: ) if self._should_try_to_reconnect(ex): + # set before waiting, so that the state is accurate for the whole of the retry delay + self._set_state(SubscriptionWorkerState.RETRYING, ex) + time.sleep(self._options.time_to_wait_before_connection_retry.total_seconds()) if self._redirect_node is None: diff --git a/ravendb/documents/subscriptions/worker_status.py b/ravendb/documents/subscriptions/worker_status.py new file mode 100644 index 00000000..f91cfb34 --- /dev/null +++ b/ravendb/documents/subscriptions/worker_status.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import datetime +from enum import Enum +from typing import Optional + + +class SubscriptionWorkerState(Enum): + """ + The stage a subscription worker is currently in. Read it through SubscriptionWorker.status. + """ + + # The worker was created but run() was not called yet, so it is not doing anything. + NOT_STARTED = "NotStarted" + # Opening a connection to the server, which includes resolving the node that owns the subscription and + # the TCP handshake. This is also the state while reconnecting after a failure. + CONNECTING = "Connecting" + # Connected to the server and waiting for it to send the next batch. The worker is healthy and idle. + WAITING_FOR_DOCUMENTS = "WaitingForDocuments" + # A batch was received and handed to the subscriber callback. The state stays here until the callback + # returned and the batch was acknowledged to the server. + PROCESSING = "Processing" + # The connection failed and the worker is going to reconnect. SubscriptionWorkerStatus.exception holds + # the failure that caused it. + RETRYING = "Retrying" + # The worker gave up: the failure it hit cannot be recovered from by reconnecting, so it stopped and the + # future returned from run() completed with that failure. SubscriptionWorkerStatus.exception holds it. + # This state is terminal. + FAULTED = "Faulted" + # The worker stopped on request, it was closed. This state is terminal. + STOPPED = "Stopped" + + +class SubscriptionWorkerStatus: + """ + An immutable snapshot of what a subscription worker was doing at a single point in time. Taking a snapshot + keeps the state and the failure that produced it consistent with each other. + + All timestamps are naive datetimes in UTC. + """ + + __slots__ = ("_state", "_exception", "_since_utc", "_failing_since_utc") + + def __init__( + self, + state: SubscriptionWorkerState, + exception: Optional[Exception], + since_utc: datetime.datetime, + failing_since_utc: Optional[datetime.datetime], + ): + self._state = state + self._exception = exception + self._since_utc = since_utc + self._failing_since_utc = failing_since_utc + + @property + def state(self) -> SubscriptionWorkerState: + """ + The stage the worker was in. + """ + return self._state + + @property + def exception(self) -> Optional[Exception]: + """ + The failure that led to the current state. Set for RETRYING and FAULTED, None otherwise. On a worker + that keeps failing this is the most recent failure, which is not necessarily the one that started + the trouble, see failing_since_utc. + """ + return self._exception + + @property + def since_utc(self) -> datetime.datetime: + """ + When the worker entered this state. Reaching the same state again does not move it, so a worker that + keeps processing batches reports the time its current batch started. A worker that cannot reach the + server alternates between CONNECTING and RETRYING, so this moves on every attempt. Use + failing_since_utc to tell how long it has been in trouble. + """ + return self._since_utc + + @property + def failing_since_utc(self) -> Optional[datetime.datetime]: + """ + When the worker last stopped being able to talk to the server, or None while it is connected. Unlike + since_utc this survives the whole reconnect cycle, so it tells how long a worker that is retrying in a + loop has been broken. It is cleared as soon as the worker is connected again, which makes a value + other than None the thing to alert on. + """ + return self._failing_since_utc + + def __str__(self) -> str: + text = f"{self._state.value} since {self._since_utc.isoformat()}" + + if self._failing_since_utc is not None: + text += f", failing since {self._failing_since_utc.isoformat()}" + + if self._exception is not None: + text += f" because of {type(self._exception).__name__}: {self._exception}" + + return text + + def __repr__(self) -> str: + return f"SubscriptionWorkerStatus({self})" diff --git a/ravendb/tests/jvm_migrated_tests/issues_tests/test_ravenDB_27175.py b/ravendb/tests/jvm_migrated_tests/issues_tests/test_ravenDB_27175.py new file mode 100644 index 00000000..43f04506 --- /dev/null +++ b/ravendb/tests/jvm_migrated_tests/issues_tests/test_ravenDB_27175.py @@ -0,0 +1,259 @@ +import time +import unittest +from threading import Event +from typing import List + +from ravendb import DocumentStore +from ravendb.documents.subscriptions.options import SubscriptionWorkerOptions +from ravendb.documents.subscriptions.worker import SubscriptionWorker +from ravendb.documents.subscriptions.worker_status import SubscriptionWorkerState, SubscriptionWorkerStatus +from ravendb.exceptions.exceptions import SubscriptionInUseException +from ravendb.infrastructure.orders import Company +from ravendb.tests.test_base import TestBase + + +class TestRavenDB27175(TestBase): + def setUp(self): + super(TestRavenDB27175, self).setUp() + self.reasonable_wait_time = 60 + + def _wait_for_state(self, worker: SubscriptionWorker, state: SubscriptionWorkerState) -> SubscriptionWorkerState: + deadline = time.monotonic() + self.reasonable_wait_time + while time.monotonic() < deadline: + if worker.status.state == state: + break + time.sleep(0.025) + return worker.status.state + + def test_can_be_notified_on_established_connection_of_an_idle_subscription(self): + name = self.store.subscriptions.create_for_class(Company) + + with self.store.subscriptions.get_subscription_worker(SubscriptionWorkerOptions(name), Company) as worker: + connected = Event() + state_when_connected: List[SubscriptionWorkerState] = [] + + def _on_established(): + state_when_connected.append(worker.status.state) + connected.set() + + worker.add_on_established_subscription_connection(_on_established) + + # no documents are ever stored here, so the event is the only indication that the worker is connected + worker.run(lambda batch: None) + + self.assertTrue(connected.wait(self.reasonable_wait_time), "initial connection wasn't established") + self.assertEqual(SubscriptionWorkerState.WAITING_FOR_DOCUMENTS, state_when_connected[0]) + + def test_status_follows_the_worker_from_connecting_through_processing(self): + with self.store.open_session() as session: + session.store(Company(name="RavenDB")) + session.save_changes() + + name = self.store.subscriptions.create_for_class(Company) + + with self.store.subscriptions.get_subscription_worker(SubscriptionWorkerOptions(name), Company) as worker: + self.assertEqual(SubscriptionWorkerState.NOT_STARTED, worker.status.state) + self.assertIsNone(worker.status.exception) + + observed: List[SubscriptionWorkerState] = [] + raised_by: List[object] = [] + + def _on_state_changed(sender, status: SubscriptionWorkerStatus): + raised_by.append(sender) + observed.append(status.state) + + worker.add_on_state_changed(_on_state_changed) + + processed = Event() + state_while_processing: List[SubscriptionWorkerState] = [] + + def _process(batch): + state_while_processing.append(worker.status.state) + processed.set() + + worker.run(_process) + + self.assertTrue(processed.wait(self.reasonable_wait_time), "the batch was never handed to the subscriber") + self.assertEqual(SubscriptionWorkerState.PROCESSING, state_while_processing[0]) + + # once the subscriber returned and the batch was acknowledged, the worker is idle again + self.assertEqual( + SubscriptionWorkerState.WAITING_FOR_DOCUMENTS, + self._wait_for_state(worker, SubscriptionWorkerState.WAITING_FOR_DOCUMENTS), + ) + + self.assertEqual( + [ + SubscriptionWorkerState.CONNECTING, + SubscriptionWorkerState.WAITING_FOR_DOCUMENTS, + SubscriptionWorkerState.PROCESSING, + ], + observed[:3], + ) + + self.assertTrue(raised_by) + for sender in raised_by: + self.assertIs(worker, sender) + + def test_status_is_faulted_with_the_failure_when_the_worker_gives_up(self): + name = self.store.subscriptions.create_for_class(Company) + + with self.store.subscriptions.get_subscription_worker(SubscriptionWorkerOptions(name), Company) as worker: + connected = Event() + worker.add_on_established_subscription_connection(connected.set) + worker.run(lambda batch: None) + self.assertTrue(connected.wait(self.reasonable_wait_time), "initial connection wasn't established") + + # the subscription is taken, which is a failure the second worker cannot recover from by reconnecting + with self.store.subscriptions.get_subscription_worker( + SubscriptionWorkerOptions(name), Company + ) as second_worker: + task = second_worker.run(lambda batch: None) + + with self.assertRaises(SubscriptionInUseException) as ctx: + task.result(self.reasonable_wait_time) + + self.assertEqual(SubscriptionWorkerState.FAULTED, second_worker.status.state) + self.assertIs(ctx.exception, second_worker.status.exception) + self.assertIsNotNone(second_worker.status.failing_since_utc) + + def test_status_is_stopped_after_close(self): + name = self.store.subscriptions.create_for_class(Company) + + worker = self.store.subscriptions.get_subscription_worker(SubscriptionWorkerOptions(name), Company) + connected = Event() + worker.add_on_established_subscription_connection(connected.set) + + worker.run(lambda batch: None) + + self.assertTrue(connected.wait(self.reasonable_wait_time), "initial connection wasn't established") + + worker.close() + + self.assertEqual(SubscriptionWorkerState.STOPPED, worker.status.state) + self.assertIsNone(worker.status.exception) + + def test_status_is_stopped_after_closing_a_worker_that_was_never_run(self): + name = self.store.subscriptions.create_for_class(Company) + + worker = self.store.subscriptions.get_subscription_worker(SubscriptionWorkerOptions(name), Company) + + self.assertEqual(SubscriptionWorkerState.NOT_STARTED, worker.status.state) + + worker.close() + + self.assertEqual(SubscriptionWorkerState.STOPPED, worker.status.state) + + +class TestSubscriptionWorkerStatusTransitions(unittest.TestCase): + """ + Covers the state bookkeeping of the worker without a server: the transitions are driven directly. + """ + + def setUp(self): + self.store = DocumentStore(["http://127.0.0.1:8080"], "db") + self.worker = SubscriptionWorker(Company, SubscriptionWorkerOptions("subscription"), False, self.store, None) + self.changes: List[SubscriptionWorkerStatus] = [] + self.worker.add_on_state_changed(lambda sender, status: self.changes.append(status)) + + def test_new_worker_is_not_started(self): + worker = SubscriptionWorker(Company, SubscriptionWorkerOptions("subscription"), False, self.store, None) + + self.assertEqual(SubscriptionWorkerState.NOT_STARTED, worker.status.state) + self.assertIsNone(worker.status.exception) + self.assertIsNone(worker.status.failing_since_utc) + self.assertIsNotNone(worker.status.since_utc) + + def test_closing_a_worker_that_was_never_run_reports_stopped(self): + senders = [] + self.worker.add_on_state_changed(lambda sender, status: senders.append(sender)) + + self.worker.close() + + self.assertEqual(SubscriptionWorkerState.STOPPED, self.worker.status.state) + self.assertEqual([SubscriptionWorkerState.STOPPED], [s.state for s in self.changes]) + self.assertEqual([self.worker], senders) + + def test_failing_since_utc_survives_the_retry_cycle_and_is_cleared_on_recovery(self): + self.worker._set_state(SubscriptionWorkerState.CONNECTING) + self.worker._set_state(SubscriptionWorkerState.WAITING_FOR_DOCUMENTS) + self.assertIsNone(self.worker.status.failing_since_utc) + + first_failure = RuntimeError("first") + self.worker._set_state(SubscriptionWorkerState.RETRYING, first_failure) + first = self.worker.status + + self.assertIs(first_failure, first.exception) + self.assertIsNotNone(first.failing_since_utc) + + time.sleep(0.01) + + # CONNECTING neither starts nor ends the failing stretch, it carries it over + self.worker._set_state(SubscriptionWorkerState.CONNECTING) + self.assertEqual(first.failing_since_utc, self.worker.status.failing_since_utc) + self.assertIsNone(self.worker.status.exception) + + second_failure = RuntimeError("second") + self.worker._set_state(SubscriptionWorkerState.RETRYING, second_failure) + later = self.worker.status + + self.assertGreater(later.since_utc, first.since_utc) + self.assertEqual(first.failing_since_utc, later.failing_since_utc) + self.assertIs(second_failure, later.exception) + + self.worker._set_state(SubscriptionWorkerState.CONNECTING) + self.assertEqual(first.failing_since_utc, self.worker.status.failing_since_utc) + + # reaching the server clears it + self.worker._set_state(SubscriptionWorkerState.WAITING_FOR_DOCUMENTS) + self.assertIsNone(self.worker.status.failing_since_utc) + self.assertIsNone(self.worker.status.exception) + + def test_reaching_the_same_state_again_does_not_raise_the_event(self): + self.worker._set_state(SubscriptionWorkerState.PROCESSING) + since = self.worker.status.since_utc + + self.worker._set_state(SubscriptionWorkerState.PROCESSING) + + self.assertEqual([SubscriptionWorkerState.PROCESSING], [s.state for s in self.changes]) + self.assertEqual(since, self.worker.status.since_utc) + + def test_a_failing_state_changed_handler_does_not_break_the_worker(self): + def _failing_handler(sender, status): + raise ValueError("handler failure") + + self.worker.remove_on_state_changed(self.worker._on_state_changed[0]) + self.worker.add_on_state_changed(_failing_handler) + self.worker.add_on_state_changed(lambda sender, status: self.changes.append(status)) + + self.worker._set_state(SubscriptionWorkerState.CONNECTING) + + self.assertEqual(SubscriptionWorkerState.CONNECTING, self.worker.status.state) + self.assertEqual([SubscriptionWorkerState.CONNECTING], [s.state for s in self.changes]) + + def test_established_connection_handlers_can_be_added_and_removed(self): + calls = [] + + def _handler(): + calls.append(1) + + self.worker.add_on_established_subscription_connection(_handler) + self.worker.invoke_on_established_subscription_connection() + self.worker.remove_on_established_subscription_connection(_handler) + self.worker.invoke_on_established_subscription_connection() + + self.assertEqual([1], calls) + + def test_status_text_contains_the_state_the_failing_time_and_the_failure(self): + self.worker._set_state(SubscriptionWorkerState.RETRYING, RuntimeError("connection lost")) + status = self.worker.status + + text = str(status) + + self.assertTrue(text.startswith(f"Retrying since {status.since_utc.isoformat()}")) + self.assertIn(f", failing since {status.failing_since_utc.isoformat()}", text) + self.assertTrue(text.endswith(" because of RuntimeError: connection lost")) + + +if __name__ == "__main__": + unittest.main()