http-api: T7090: Implement background configure operations for REST API

Large config commits (`service config-sync`) can block the REST API request
path and sometimes must be deferred (e.g., when changing `service https`).

This commit introduces an in-memory background operation manager
that queues (FIFO) full configure operations (commands + commit/commit-confirm)
as single jobs, tracks status/result, and exposes active operations
via `/retrieve/background-operations`.

3 files changed, 306 insertions, 29 deletions

diff --git a/src/helpers/vyos_config_sync.py b/src/helpers/vyos_config_sync.py
index 80bfb6d17..f0650c5fc 100755
--- a/src/helpers/vyos_config_sync.py
+++ b/src/helpers/vyos_config_sync.py
@@ -40,9 +40,12 @@ logger.name = os.path.basename(__file__)
 API_HEADERS = {'Content-Type': 'application/json'}
 
 
-def post_request(url: str,
-                 data: str,
-                 headers: Dict[str, str]) -> requests.Response:
+def post_request(
+    url: str,
+    data: str,
+    params: Dict[str, Any],
+    headers: Dict[str, str],
+) -> requests.Response:
     """Sends a POST request to the specified URL
 
     Args:
@@ -54,11 +57,14 @@ def post_request(url: str,
         requests.Response: The response object representing the server's response to the request
     """
 
-    response = requests.post(url,
-                             data=data,
-                             headers=headers,
-                             verify=False,
-                             timeout=timeout)
+    response = requests.post(
+        url,
+        data=data,
+        params=params,
+        headers=headers,
+        verify=False,
+        timeout=timeout,
+    )
     return response
 
 
@@ -116,6 +122,10 @@ def set_remote_config(
     urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 
     url = f'https://{address}:{port}/configure-section'
+    params = {
+        # Ask the remote API to perform the configure and commit workflow asynchronously
+        'in_background': True,
+    }
     data = json.dumps({
         'op': op,
         'mask': mask,
@@ -124,7 +134,7 @@ def set_remote_config(
     })
 
     try:
-        config = post_request(url, data, headers)
+        config = post_request(url, data, params, headers)
         return config.json()
     except requests.exceptions.RequestException as e:
         print(f"An error occurred: {e}")
diff --git a/src/services/api/background.py b/src/services/api/background.py
new file mode 100644
index 000000000..2b25af307
--- /dev/null
+++ b/src/services/api/background.py
@@ -0,0 +1,179 @@
+# Copyright VyOS maintainers and contributors <maintainers@vyos.io>
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this library.  If not, see <http://www.gnu.org/licenses/>.
+
+import time
+import functools
+from collections import deque
+from enum import Enum
+from threading import Lock
+from typing import Any
+from typing import Callable
+from typing import Optional
+from uuid import uuid4
+
+from fastapi import BackgroundTasks
+from pydantic import BaseModel
+from pydantic import StrictStr
+from pydantic import StrictInt
+
+
+def _ts():
+    """Return current Unix timestamp (seconds since epoch)"""
+    return int(time.time())
+
+
+class BackgroundOpStatus(str, Enum):
+    queued = 'queued'
+    running = 'running'
+    succeeded = 'succeeded'
+    failed = 'failed'
+
+    @property
+    def is_completed(self):
+        """True if the operation is in a terminal state (succeeded/failed)"""
+        return self in (BackgroundOpStatus.succeeded, BackgroundOpStatus.failed)
+
+
+class BackgroundOpRecord(BaseModel):
+    """Metadata and outcome for a single background operation"""
+
+    op_id: StrictStr
+    created_at: StrictInt
+    started_at: Optional[StrictInt] = None
+    finished_at: Optional[StrictInt] = None
+    status: BackgroundOpStatus = BackgroundOpStatus.queued
+    result: Optional[Any] = None
+    error: Optional[StrictStr] = None
+
+
+class BackgroundOpError(Exception):
+    """Raised when a background operation cannot be enqueued/executed"""
+
+    pass
+
+
+class BackgroundOpManager:
+    """
+    In-memory FIFO operation queue.
+
+    Uses BackgroundTasks to schedule a `drain()` call after the response,
+    so `enqueue()` is fast and non-blocking for the client.
+    """
+
+    DEFAULT_MAX_QUEUE_SIZE = 128
+
+    def __init__(self, max_queue_size: int = DEFAULT_MAX_QUEUE_SIZE):
+        # max number of queued (pending) operations allowed at a time
+        self._max_queue_size = max_queue_size
+
+        # FIFO queue of operation IDs waiting to be executed
+        self._queue = deque()
+        self._jobs = {}
+        self._workers = {}
+
+        # protects _queue/_jobs/_workers/_drain_scheduled from concurrent access
+        self._mx = Lock()
+
+        # whether a drain task has already been scheduled via BackgroundTasks
+        self._drain_scheduled = False
+
+    def enqueue(
+        self,
+        background_tasks: BackgroundTasks,
+        func: Callable,
+        *args,
+        **kwargs,
+    ) -> BackgroundOpRecord:
+        """Enqueue a function for background execution and return its record"""
+
+        assert isinstance(background_tasks, BackgroundTasks)
+        assert callable(func), '`func` argument should be function or lambda'
+
+        with self._mx:
+            if len(self._queue) >= self._max_queue_size:
+                raise BackgroundOpError(
+                    f'Background operation queue is full ({self._max_queue_size})'
+                )
+
+            op_id = str(uuid4())
+            record = BackgroundOpRecord(op_id=op_id, created_at=_ts())
+
+            self._jobs[op_id] = record
+            # store the callable for later execution (outside the lock)
+            self._workers[op_id] = functools.partial(func, *args, **kwargs)
+            self._queue.append(op_id)
+
+            if not self._drain_scheduled:
+                # schedule a single drain() call after the current response
+                background_tasks.add_task(self.drain)
+                self._drain_scheduled = True
+
+            # Best-effort pruning: keep history bounded by dropping oldest completed records
+            if len(self._jobs) > self._max_queue_size:
+                oldest = min(self._jobs.values(), key=lambda record: record.created_at)
+                if oldest.status.is_completed:
+                    del self._jobs[oldest.op_id]
+
+        return record
+
+    def drain(self):
+        """Run queued operations sequentially until the queue is empty"""
+
+        while True:
+            with self._mx:
+                if not self._queue:
+                    # allow future enqueue() calls to schedule the next drain()
+                    self._drain_scheduled = False
+                    return
+
+                op_id = self._queue.popleft()
+                record = self._jobs[op_id]
+                func = self._workers.pop(op_id)
+
+                record.status = BackgroundOpStatus.running
+                record.started_at = _ts()
+
+            # execute outside the lock to avoid blocking enqueues/status reads
+            result = error = status = None
+            try:
+                result = func()
+            except Exception as e:  # noqa: BLE001
+                status = BackgroundOpStatus.failed
+                error = str(e)
+            else:
+                status = BackgroundOpStatus.succeeded
+
+            with self._mx:
+                record.result = result
+                record.error = error
+                record.status = status
+                record.finished_at = _ts()
+
+    def get_record(self, op_id: str) -> BackgroundOpRecord | None:
+        """Return a deep copy of a single record"""
+
+        with self._mx:
+            record = self._jobs.get(op_id)
+            return record.copy(deep=True) if record else None
+
+    def get_records(self) -> list:
+        """Return deep copies of all records, sorted oldest-first by created_at"""
+
+        with self._mx:
+            records = [record.copy(deep=True) for record in self._jobs.values()]
+
+        # stable-ish ordering (oldest first)
+        records.sort(key=lambda record: record.created_at)
+        return records
diff --git a/src/services/api/rest/routers.py b/src/services/api/rest/routers.py
index 438b3a23e..c8102c980 100644
--- a/src/services/api/rest/routers.py
+++ b/src/services/api/rest/routers.py
@@ -28,6 +28,7 @@ from typing import Callable
 from typing import TYPE_CHECKING
 
 from fastapi import Depends
+from fastapi import Query
 from fastapi import Request
 from fastapi import Response
 from fastapi import HTTPException
@@ -46,6 +47,8 @@ from vyos.configtree import ConfigTree
 from vyos.configdiff import get_config_diff
 from vyos.configsession import ConfigSessionError
 
+from ..background import BackgroundOpManager
+from ..background import BackgroundOpError
 from ..session import SessionState
 from .models import success
 from .models import error
@@ -230,7 +233,7 @@ class MultipartRequest(Request):
                                 400,
                                 f"Malformed command '{0}': 'path' field must be a list of strings",
                             )
-                    if endpoint in ('/configure'):
+                    if endpoint in ('/configure',):
                         if not c['path']:
                             self.form_err = (
                                 400,
@@ -241,7 +244,7 @@ class MultipartRequest(Request):
                                 400,
                                 f"Malformed command '{c}': 'value' field must be a string",
                             )
-                    if endpoint in ('/configure-section'):
+                    if endpoint in ('/configure-section',):
                         if 'section' not in c and 'config' not in c:
                             self.form_err = (
                                 400,
@@ -293,6 +296,10 @@ router = APIRouter(
 self_ref_msg = 'Requested HTTP API server configuration change; commit will be called in the background'
 
 
+# Global background-op manager used by the REST API to run long config commits after the response
+background_op_manager = BackgroundOpManager()
+
+
 def call_commit(s: SessionState):
     try:
         s.session.commit()
@@ -342,7 +349,7 @@ def run_commit_confirm(s: SessionState):
         del env['IN_COMMIT_CONFIRM']
 
 
-async def _configure_op(
+def _execute_configure_op(
     data: Union[
         ConfirmModel,
         ConfigureModel,
@@ -351,12 +358,15 @@ async def _configure_op(
         ConfigSectionListModel,
         ConfigSectionTreeModel,
     ],
-    _request: Request,
-    background_tasks: BackgroundTasks,
+    background_tasks: BackgroundTasks | None = None,
 ):
     # pylint: disable=too-many-branches,too-many-locals,too-many-nested-blocks,too-many-statements
     # pylint: disable=consider-using-with
 
+    # True when invoked by the background operation
+    # runner (no FastAPI BackgroundTasks context passed in)
+    is_background_job = background_tasks is None
+
     state = SessionState()
     session = state.session
     env = session.get_session_env()
@@ -385,6 +395,8 @@ async def _configure_op(
     try:
         for c in data:
             op = c.op
+            op_error = ConfigSessionError(f"'{op}' is not a valid operation")
+
             if not isinstance(c, (ConfirmModel, BaseConfigSectionTreeModel)):
                 path = c.path
 
@@ -392,7 +404,7 @@ async def _configure_op(
                 if op == 'confirm':
                     msg = session.confirm()
                 else:
-                    raise ConfigSessionError(f"'{op}' is not a valid operation")
+                    raise op_error
 
             elif isinstance(c, BaseConfigureModel):
                 if c.value:
@@ -422,7 +434,7 @@ async def _configure_op(
                 elif op == 'comment':
                     session.comment(path, value=value)
                 else:
-                    raise ConfigSessionError(f"'{op}' is not a valid operation")
+                    raise op_error
 
             elif isinstance(c, BaseConfigSectionModel):
                 if op == 'set':
@@ -430,7 +442,7 @@ async def _configure_op(
                 elif op == 'load':
                     session.load_section(path, section)
                 else:
-                    raise ConfigSessionError(f"'{op}' is not a valid operation")
+                    raise op_error
 
             elif isinstance(c, BaseConfigSectionTreeModel):
                 if op == 'set':
@@ -438,7 +450,7 @@ async def _configure_op(
                 elif op == 'load':
                     session.load_section_tree(mask, config)
                 else:
-                    raise ConfigSessionError(f"'{op}' is not a valid operation")
+                    raise op_error
         # end for
 
         config = Config(session_env=env)
@@ -448,20 +460,29 @@ async def _configure_op(
 
         if not d.is_node_changed(['service', 'https']):
             if confirm_time:
-                out, err = await run_in_threadpool(run_commit_confirm, state)
+                out, err = run_commit_confirm(state)
                 if err:
                     raise err
                 msg = msg + out if msg else out
             else:
-                out, err = await run_in_threadpool(run_commit, state)
+                out, err = run_commit(state)
                 if err:
                     raise err
                 msg = msg + out if msg else out
         else:
-            if confirm_time:
-                background_tasks.add_task(call_commit_confirm, state)
+            if is_background_job:
+                # If already running as a background job, commit synchronously here
+                if confirm_time:
+                    call_commit_confirm(state)
+                else:
+                    call_commit(state)
             else:
-                background_tasks.add_task(call_commit, state)
+                # Otherwise schedule the commit to run after the HTTP response
+                if confirm_time:
+                    background_tasks.add_task(call_commit_confirm, state)
+                else:
+                    background_tasks.add_task(call_commit, state)
+
             out = self_ref_msg
             msg = msg + out if msg else out
 
@@ -484,12 +505,62 @@ async def _configure_op(
             del env['IN_COMMIT_CONFIRM']
         lock.release()
 
+    # Background jobs return raw success text or raise on failure;
+    # the API wrapper formats HTTP responses and returns it
+    if is_background_job:
+        if status == 200:
+            return msg
+        else:
+            raise RuntimeError(error_msg)
+
     if status != 200:
         return error(status, error_msg)
 
     return success(msg)
 
 
+async def _configure_op(
+    data: Union[
+        ConfirmModel,
+        ConfigureModel,
+        ConfigureListModel,
+        ConfigSectionModel,
+        ConfigSectionListModel,
+        ConfigSectionTreeModel,
+    ],
+    background_tasks: BackgroundTasks,
+    in_background: bool = False,
+):
+    """
+    API wrapper for configure operations.
+
+    If `in_background=True`: enqueue the whole configure
+    workflow and return an operation record immediately.
+    Otherwise: run the configure workflow in a threadpool
+    and return the normal API response.
+    """
+
+    if in_background:
+        try:
+            # Enqueue and return an operation handle that
+            # can be polled via `/retrieve/background-operations`
+            record = background_op_manager.enqueue(
+                background_tasks,
+                _execute_configure_op,
+                data,
+            )
+        except BackgroundOpError as e:
+            return error(500, str(e))
+
+        return success({'operation': record.model_dump()})
+
+    return await run_in_threadpool(
+        _execute_configure_op,
+        data,
+        background_tasks=background_tasks,
+    )
+
+
 def create_path_import_pki_no_prompt(path):
     correct_paths = ['ca', 'certificate', 'key-pair']
     if path[1] not in correct_paths:
@@ -504,10 +575,9 @@ async def configure_op(
     data: Union[ConfigureModel, ConfigureListModel, ConfirmModel],
     request: Request,
     background_tasks: BackgroundTasks,
+    in_background: bool = Query(False),
 ):
-    out = await _configure_op(data, request, background_tasks)
-
-    return out
+    return await _configure_op(data, background_tasks, in_background)
 
 
 @router.post('/configure-section')
@@ -515,10 +585,9 @@ async def configure_section_op(
     data: Union[ConfigSectionModel, ConfigSectionListModel, ConfigSectionTreeModel],
     request: Request,
     background_tasks: BackgroundTasks,
+    in_background: bool = Query(False),
 ):
-    out = await _configure_op(data, request, background_tasks)
-
-    return out
+    return await _configure_op(data, background_tasks, in_background)
 
 
 @router.post('/retrieve')
@@ -565,6 +634,25 @@ async def retrieve_op(data: RetrieveModel):
     return success(res)
 
 
+@router.post('/retrieve/background-operations')
+async def retrieve_background_operations(
+    op_id: str = Query(None),
+):
+    if op_id:
+        # Return only that record
+        record = background_op_manager.get_record(op_id)
+        records = [record] if record else []
+    else:
+        # Return the full in-memory operation history (oldest first)
+        records = background_op_manager.get_records()
+
+    result = {
+        'operations': [record.model_dump() for record in records],
+    }
+
+    return success(result)
+
+
 @router.post('/config-file')
 async def config_file_op(data: ConfigFileModel, background_tasks: BackgroundTasks):
     state = SessionState()