PATCH /ports/{id}/value Timeout

qtoggleserver/core/api/__init__.py

@@ -95,7 +95,7 @@ def wrapper(request_handler: APIHandler, *args, **kwargs) -> Any:
                 else:
                     raise APIError(403, "forbidden", required_level=ACCESS_LEVEL_MAPPING.get(access_level))
 
-            request = APIRequest(request_handler)
+            request = request_handler if isinstance(request_handler, APIRequest) else APIRequest(request_handler)
 
             return func(request, *args, **kwargs)
 

qtoggleserver/core/api/funcs/ports.py

@@ -28,6 +28,9 @@
 from qtoggleserver.utils import json as json_utils
 
 
+MAX_VALUE_TIMEOUT = 3600  # seconds
+
+
 async def add_virtual_port(attrs: GenericJSONDict) -> core_ports.BasePort:
     id_ = attrs["id"]
     type_ = attrs["type"]
@@ -351,10 +354,23 @@ async def get_port_value(request: core_api.APIRequest, port_id: str) -> Nullable
 
 @core_api.api_call(core_api.ACCESS_LEVEL_NORMAL)
 async def patch_port_value(request: core_api.APIRequest, port_id: str, params: PortValue) -> None:
+    request_time = time.time()
+
     port = core_ports.get(port_id)
     if port is None:
         raise core_api.APIError(404, "no-such-port")
 
+    timeout_str = request.query.get("timeout")
+    timeout = 0
+    if timeout_str is not None:
+        try:
+            timeout = float(timeout_str)
+        except ValueError:
+            raise core_api.APIError(400, "invalid-field", field="timeout") from None
+
+        if timeout < 0 or timeout > MAX_VALUE_TIMEOUT:
+            raise core_api.APIError(400, "invalid-field", field="timeout")
+
     try:
         core_api_schema.validate(params, await port.get_value_schema())
     except core_api.APIError:
@@ -387,6 +403,11 @@ async def patch_port_value(request: core_api.APIRequest, port_id: str, params: P
         # Transform any unhandled exception into APIError(500)
         raise core_api.APIError(500, "unexpected-error", message=str(e)) from e
 
+    if timeout:
+        remaining = timeout - (time.time() - request_time)
+        if not await port.wait_for_read_value(value, timeout=remaining):
+            raise core_api.APIError(504, "value-timeout")
+
 
 @core_api.api_call(core_api.ACCESS_LEVEL_NORMAL)
 async def patch_port_sequence(request: core_api.APIRequest, port_id: str, params: GenericJSONDict) -> None:

qtoggleserver/core/ports.py

@@ -249,6 +249,7 @@ def __init__(self, port_id: str) -> None:
 
         self._last_read_value: tuple[NullablePortValue, int] | None = None
         self._read_value_lock = asyncio.Lock()
+        self._value_match_waiters: list[tuple[NullablePortValue, asyncio.Future]] = []
 
         # Value that's currently being written
         self._writing_value: NullablePortValue = None
@@ -672,6 +673,32 @@ def get_last_read_value(self) -> NullablePortValue:
     def set_last_read_value(self, value: NullablePortValue) -> None:
         self._last_read_value = value, int(time.time() * 1000)
 
+        for target_value, future in self._value_match_waiters:
+            if not future.done() and value == target_value:
+                future.set_result(None)
+
+    async def wait_for_read_value(self, value: NullablePortValue, timeout: float) -> bool:
+        """Wait until the port's last read value becomes equal to `value`, but no longer than `timeout` seconds.
+        Returns `True` as soon as a matching value is read (right away if it already matches), or `False` if
+        `timeout` seconds elapse without a match."""
+
+        if self.get_last_read_value() == value:
+            return True
+
+        if timeout <= 0:
+            return False
+
+        future = asyncio.get_running_loop().create_future()
+        waiter = (value, future)
+        self._value_match_waiters.append(waiter)
+        try:
+            await asyncio.wait_for(future, timeout=timeout)
+            return True
+        except TimeoutError:
+            return False
+        finally:
+            self._value_match_waiters.remove(waiter)
+
     async def read_transformed_value(self) -> NullablePortValue:
         value = None
         async with self._read_value_lock:

qtoggleserver/frontend/js/api/constants.js

@@ -218,6 +218,11 @@ export const KNOWN_ERRORS = [
         code: 'device-timeout',
         pretty: gettext('Timeout waiting for a response from the device.')
     },
+    {
+        status: 504,
+        code: 'value-timeout',
+        pretty: gettext('Timeout waiting for value to take effect.')
+    },
 
     /* Other errors */
     {

qtoggleserver/frontend/js/api/ports.js

@@ -120,11 +120,12 @@ export function getPortValue(id) {
  * @alias qtoggle.api.ports.patchPortValue
  * @param {String} id the port identifier
  * @param {Boolean|Number} value the new port value
+ * @param {?Number} [confirmTimeout] optional confirmation timeout, in seconds
  * @param {Number} [expectEventTimeout] optional timeout within which a corresponding event will be expected, in
  * milliseconds
  * @returns {Promise}
  */
-export function patchPortValue(id, value, expectEventTimeout = null) {
+export function patchPortValue(id, value, confirmTimeout = null, expectEventTimeout = null) {
     let port = Cache.getPort(id)
     let handle = null
 
@@ -142,12 +143,22 @@ export function patchPortValue(id, value, expectEventTimeout = null) {
         }, expectEventTimeout)
     }
 
+    let query = null
+    let timeout = APIConstants.LONG_SERVER_TIMEOUT
+    if (confirmTimeout != null) {
+        query = {timeout: confirmTimeout}
+
+        /* Make sure the client waits at least as long as the device is expected to, plus some margin */
+        timeout = Math.max(timeout, confirmTimeout + APIConstants.DEFAULT_SERVER_TIMEOUT)
+    }
+
     return BaseAPI.apiCall({
         method: 'PATCH',
         path: `/ports/${id}/value`,
         data: value,
+        query: query,
         expectedHandle: handle,
-        timeout: APIConstants.LONG_SERVER_TIMEOUT
+        timeout: timeout
     })
 }
 

qtoggleserver/frontend/js/dashboard/widgets/widget.js

@@ -2,8 +2,6 @@
 import $                     from '$qui/lib/jquery.module.js'
 import Logger                from '$qui/lib/logger.module.js'
 
-import ConditionVariable    from '$qui/base/condition-variable.js'
-import {TimeoutError}       from '$qui/base/errors.js'
 import {gettext}            from '$qui/base/i18n.js'
 import {mix}                from '$qui/base/mixwith.js'
 import StockIcon            from '$qui/icons/stock-icon.js'
@@ -104,10 +102,6 @@ class Widget extends mix().with(ViewMixin) {
             this._height = this.constructor.height
         }
 
-        this._valueChangeWaitPortId = null
-        this._whenValueChange = null
-        this._valueChangeTimeoutHandle = null
-
         this.logger = Logger.get(this.makeLogName())
     }
 
@@ -1129,14 +1123,6 @@ class Widget extends mix().with(ViewMixin) {
     /* Ports and port values */
 
     handlePortValueChange(portId, value) {
-        if (this._valueChangeWaitPortId === portId && this._whenValueChange) {
-            this._whenValueChange.fulfill()
-            this._whenValueChange = null
-            this._valueChangeWaitPortId = null
-            clearTimeout(this._valueChangeTimeoutHandle)
-            this._valueChangeTimeoutHandle = null
-        }
-
         this.onPortValueChange(portId, value)
     }
 
@@ -1168,46 +1154,13 @@ class Widget extends mix().with(ViewMixin) {
      * @param {String} portId the id of the port whose value will be set
      * @param {Number|Boolean} value the new port value
      * @param {Number} [timeout] how long to wait for new port value to take effect (seconds, defaults to
-     * {@link qtoggle.api.constants.DEFAULT_SERVER_TIMEOUT})
+     * {@link qtoggle.api.constants.DEFAULT_SERVER_TIMEOUT}); pass `0` to not wait for confirmation at all
      * @returns {Promise}
      */
     setPortValue(portId, value, timeout = APIConstants.DEFAULT_SERVER_TIMEOUT) {
         this.setProgress()
 
-        let prevValue = this.getPortValue(portId)
-
-        if (this._whenValueChange) {
-            this._whenValueChange.fulfill()
-            this._whenValueChange = null
-            this._valueChangeWaitPortId = null
-            clearTimeout(this._valueChangeTimeoutHandle)
-            this._valueChangeTimeoutHandle = null
-        }
-
-        this._whenValueChange = new ConditionVariable()
-        this._valueChangeWaitPortId = portId
-
-        return PortsAPI.patchPortValue(portId, value).then(function () {
-
-            if (value === prevValue || timeout === 0) {
-                return /* Value was already set or we're not interested in waiting */
-            }
-
-            this._valueChangeTimeoutHandle = setTimeout(function () {
-                /* Cancel waiting after timeout */
-                if (this._whenValueChange) {
-                    let msg = gettext('Timeout waiting for value to take effect.')
-                    this._whenValueChange.cancel(new TimeoutError(msg))
-                    this._whenValueChange = null
-                    this._valueChangeWaitPortId = null
-                    this._valueChangeTimeoutHandle = null
-                }
-
-            }.bind(this), timeout * 1000)
-
-            return this._whenValueChange
-
-        }.bind(this)).then(function () {
+        return PortsAPI.patchPortValue(portId, value, timeout).then(function () {
 
             this.clearProgress()
 

tests/unit/qtoggleserver/core/api/test_funcs_ports.py

@@ -1,3 +1,5 @@
+import asyncio
+
 import pytest
 
 from qtoggleserver.conf import settings
@@ -179,3 +181,130 @@ async def test_read_only_port(self, mock_api_request_maker, mock_num_port1, mock
         with pytest.raises(core_api.APIError, match="read-only-port") as exc_info:
             await ports_api_funcs.patch_port_value(request, "nid1", 100)
         assert exc_info.value.status == 400
+
+
+class TestPatchPortValueTimeout:
+    @pytest.fixture(autouse=True)
+    def mock_slaves(self, mocker) -> None:
+        mocker.patch("qtoggleserver.slaves.devices.get_all", return_value=[])
+
+    async def test_non_integer_timeout_raises_invalid_field(
+        self, mock_api_request_maker, mock_num_port1, mock_persist_driver
+    ) -> None:
+        mock_num_port1.set_writable(True)
+
+        request = mock_api_request_maker(
+            "PATCH", "/ports/nid1/value", access_level=core_api.ACCESS_LEVEL_NORMAL, query={"timeout": "abc"}
+        )
+        with pytest.raises(core_api.APIError, match="invalid-field") as exc_info:
+            await ports_api_funcs.patch_port_value(request, "nid1", 100)
+        assert exc_info.value.status == 400
+        assert exc_info.value.params["field"] == "timeout"
+
+    async def test_negative_timeout_raises_invalid_field(
+        self, mock_api_request_maker, mock_num_port1, mock_persist_driver
+    ) -> None:
+        mock_num_port1.set_writable(True)
+
+        request = mock_api_request_maker(
+            "PATCH", "/ports/nid1/value", access_level=core_api.ACCESS_LEVEL_NORMAL, query={"timeout": "-1"}
+        )
+        with pytest.raises(core_api.APIError, match="invalid-field") as exc_info:
+            await ports_api_funcs.patch_port_value(request, "nid1", 100)
+        assert exc_info.value.status == 400
+        assert exc_info.value.params["field"] == "timeout"
+
+    async def test_too_large_timeout_raises_invalid_field(
+        self, mock_api_request_maker, mock_num_port1, mock_persist_driver
+    ) -> None:
+        mock_num_port1.set_writable(True)
+
+        request = mock_api_request_maker(
+            "PATCH", "/ports/nid1/value", access_level=core_api.ACCESS_LEVEL_NORMAL, query={"timeout": "3601"}
+        )
+        with pytest.raises(core_api.APIError, match="invalid-field") as exc_info:
+            await ports_api_funcs.patch_port_value(request, "nid1", 100)
+        assert exc_info.value.status == 400
+        assert exc_info.value.params["field"] == "timeout"
+
+    async def test_no_timeout_does_not_wait_for_confirmation(
+        self, mock_api_request_maker, mock_num_port1, mock_persist_driver, mocker
+    ) -> None:
+        """Without a `timeout` query argument, the request must be responded as soon as the write completes,
+        without waiting for the read value to catch up."""
+
+        mock_num_port1.set_writable(True)
+        spy = mocker.spy(mock_num_port1, "wait_for_read_value")
+
+        request = mock_api_request_maker("PATCH", "/ports/nid1/value", access_level=core_api.ACCESS_LEVEL_NORMAL)
+        result = await ports_api_funcs.patch_port_value(request, "nid1", 100)
+
+        assert result is None
+        spy.assert_not_called()
+
+    async def test_returns_immediately_if_value_already_matches(
+        self, mock_api_request_maker, mock_num_port1, mock_persist_driver, mocker
+    ) -> None:
+        mock_num_port1.set_writable(True)
+        mock_num_port1.set_last_read_value(100)
+        spy = mocker.spy(mock_num_port1, "wait_for_read_value")
+
+        request = mock_api_request_maker(
+            "PATCH", "/ports/nid1/value", access_level=core_api.ACCESS_LEVEL_NORMAL, query={"timeout": "5"}
+        )
+        result = await ports_api_funcs.patch_port_value(request, "nid1", 100)
+
+        assert result is None
+        spy.assert_called_once()
+
+    async def test_waits_and_succeeds_once_matching_value_is_read(
+        self, mock_api_request_maker, mock_num_port1, mock_persist_driver
+    ) -> None:
+        mock_num_port1.set_writable(True)
+        mock_num_port1.set_last_read_value(0)
+
+        async def delayed_read() -> None:
+            await asyncio.sleep(0.01)
+            mock_num_port1.set_last_read_value(100)
+
+        task = asyncio.create_task(delayed_read())
+
+        request = mock_api_request_maker(
+            "PATCH", "/ports/nid1/value", access_level=core_api.ACCESS_LEVEL_NORMAL, query={"timeout": "5"}
+        )
+        try:
+            result = await ports_api_funcs.patch_port_value(request, "nid1", 100)
+        finally:
+            await task
+
+        assert result is None
+
+    async def test_raises_value_timeout_if_value_never_matches(
+        self, mock_api_request_maker, mock_num_port1, mock_persist_driver
+    ) -> None:
+        mock_num_port1.set_writable(True)
+        mock_num_port1.set_last_read_value(0)
+
+        request = mock_api_request_maker(
+            "PATCH", "/ports/nid1/value", access_level=core_api.ACCESS_LEVEL_NORMAL, query={"timeout": "0.02"}
+        )
+        with pytest.raises(core_api.APIError, match="value-timeout") as exc_info:
+            await ports_api_funcs.patch_port_value(request, "nid1", 100)
+        assert exc_info.value.status == 504
+
+    async def test_erroneous_write_skips_confirmation(
+        self, mock_api_request_maker, mock_num_port1, mock_persist_driver, mocker
+    ) -> None:
+        """An erroneous write must be responded as usual, without waiting for any confirmation."""
+
+        mock_num_port1.set_writable(True)
+        mocker.patch.object(mock_num_port1, "push_write_and_wait", side_effect=core_ports.PortTimeout())
+        spy = mocker.spy(mock_num_port1, "wait_for_read_value")
+
+        request = mock_api_request_maker(
+            "PATCH", "/ports/nid1/value", access_level=core_api.ACCESS_LEVEL_NORMAL, query={"timeout": "5"}
+        )
+        with pytest.raises(core_api.APIError, match="port-timeout") as exc_info:
+            await ports_api_funcs.patch_port_value(request, "nid1", 100)
+        assert exc_info.value.status == 504
+        spy.assert_not_called()