mirror of
https://github.com/navidrome/navidrome.git
synced 2026-03-04 06:35:52 +00:00
* feat(plugins): define TaskQueue host service interface Add the TaskQueueService interface with CreateQueue, Enqueue, GetTaskStatus, and CancelTask methods plus QueueConfig struct. * feat(plugins): define TaskWorker capability for task execution callbacks * feat(plugins): add taskqueue permission to manifest schema Add TaskQueuePermission with maxConcurrency option. * feat(plugins): implement TaskQueue service with SQLite persistence and workers Per-plugin SQLite database with queues and tasks tables. Worker goroutines dequeue tasks and invoke nd_task_execute callback. Exponential backoff retries, rate limiting via delayMs, automatic cleanup of terminal tasks. * feat(plugins): require TaskWorker capability for taskqueue permission * feat(plugins): register TaskQueue host service in manager * feat(plugins): add test-taskqueue plugin for integration testing * feat(plugins): add integration tests for TaskQueue host service * docs: document TaskQueue module for persistent task queues Signed-off-by: Deluan <deluan@navidrome.org> * fix(plugins): harden TaskQueue host service with validation and safety improvements Add input validation (queue name length, payload size limits), extract status string constants to eliminate raw SQL literals, make CreateQueue idempotent via upsert for crash recovery, fix RetentionMs default check for negative values, cap exponential backoff at 1 hour to prevent overflow, and replace manual mutex-based delay enforcement with rate.Limiter from golang.org/x/time/rate for correct concurrent worker serialization. * refactor(plugins): remove capability check for TaskWorker in TaskQueue host service Signed-off-by: Deluan <deluan@navidrome.org> * fix(plugins): use context-aware database execution in TaskQueue host service Signed-off-by: Deluan <deluan@navidrome.org> * refactor(plugins): streamline task queue configuration and error handling Signed-off-by: Deluan <deluan@navidrome.org> * feat(plugins): increase maxConcurrency for task queue and handle budget exhaustion Signed-off-by: Deluan <deluan@navidrome.org> * refactor(plugins): simplify goroutine management in task queue service Signed-off-by: Deluan <deluan@navidrome.org> * feat(plugins): update TaskWorker interface to return status messages and refactor task queue service Signed-off-by: Deluan <deluan@navidrome.org> * feat(plugins): add ClearQueue function to remove pending tasks from a specified queue Signed-off-by: Deluan <deluan@navidrome.org> * refactor(plugins): use migrateDB for task queue schema and fix constant name collision Replaced the raw db.Exec call in createTaskQueueSchema with migrateDB, matching the pattern used by createKVStoreSchema. This enables version-tracked schema migrations via SQLite's PRAGMA user_version, allowing future schema changes to be appended incrementally. Also renamed cleanupInterval to taskCleanupInterval to resolve a redeclaration conflict with host_kvstore.go. * regenerate PDKs Signed-off-by: Deluan <deluan@navidrome.org> --------- Signed-off-by: Deluan <deluan@navidrome.org>
189 lines
5.4 KiB
Python
189 lines
5.4 KiB
Python
# Code generated by ndpgen. DO NOT EDIT.
|
|
#
|
|
# This file contains client wrappers for the Task host service.
|
|
# It is intended for use in Navidrome plugins built with extism-py.
|
|
#
|
|
# IMPORTANT: Due to a limitation in extism-py, you cannot import this file directly.
|
|
# The @extism.import_fn decorators are only detected when defined in the plugin's
|
|
# main __init__.py file. Copy the needed functions from this file into your plugin.
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import extism
|
|
import json
|
|
import base64
|
|
|
|
|
|
class HostFunctionError(Exception):
|
|
"""Raised when a host function returns an error."""
|
|
pass
|
|
|
|
|
|
@extism.import_fn("extism:host/user", "task_createqueue")
|
|
def _task_createqueue(offset: int) -> int:
|
|
"""Raw host function - do not call directly."""
|
|
...
|
|
|
|
|
|
@extism.import_fn("extism:host/user", "task_enqueue")
|
|
def _task_enqueue(offset: int) -> int:
|
|
"""Raw host function - do not call directly."""
|
|
...
|
|
|
|
|
|
@extism.import_fn("extism:host/user", "task_get")
|
|
def _task_get(offset: int) -> int:
|
|
"""Raw host function - do not call directly."""
|
|
...
|
|
|
|
|
|
@extism.import_fn("extism:host/user", "task_cancel")
|
|
def _task_cancel(offset: int) -> int:
|
|
"""Raw host function - do not call directly."""
|
|
...
|
|
|
|
|
|
@extism.import_fn("extism:host/user", "task_clearqueue")
|
|
def _task_clearqueue(offset: int) -> int:
|
|
"""Raw host function - do not call directly."""
|
|
...
|
|
|
|
|
|
def task_create_queue(name: str, config: Any) -> None:
|
|
"""CreateQueue creates a named task queue with the given configuration.
|
|
Zero-value fields in config use sensible defaults.
|
|
If a queue with the same name already exists, returns an error.
|
|
On startup, this also recovers any stale "running" tasks from a previous crash.
|
|
|
|
Args:
|
|
name: str parameter.
|
|
config: Any parameter.
|
|
|
|
Raises:
|
|
HostFunctionError: If the host function returns an error.
|
|
"""
|
|
request = {
|
|
"name": name,
|
|
"config": config,
|
|
}
|
|
request_bytes = json.dumps(request).encode("utf-8")
|
|
request_mem = extism.memory.alloc(request_bytes)
|
|
response_offset = _task_createqueue(request_mem.offset)
|
|
response_mem = extism.memory.find(response_offset)
|
|
response = json.loads(extism.memory.string(response_mem))
|
|
|
|
if response.get("error"):
|
|
raise HostFunctionError(response["error"])
|
|
|
|
|
|
|
|
def task_enqueue(queue_name: str, payload: bytes) -> str:
|
|
"""Enqueue adds a task to the named queue. Returns the task ID.
|
|
payload is opaque bytes passed back to the plugin on execution.
|
|
|
|
Args:
|
|
queue_name: str parameter.
|
|
payload: bytes parameter.
|
|
|
|
Returns:
|
|
str: The result value.
|
|
|
|
Raises:
|
|
HostFunctionError: If the host function returns an error.
|
|
"""
|
|
request = {
|
|
"queueName": queue_name,
|
|
"payload": base64.b64encode(payload).decode("ascii"),
|
|
}
|
|
request_bytes = json.dumps(request).encode("utf-8")
|
|
request_mem = extism.memory.alloc(request_bytes)
|
|
response_offset = _task_enqueue(request_mem.offset)
|
|
response_mem = extism.memory.find(response_offset)
|
|
response = json.loads(extism.memory.string(response_mem))
|
|
|
|
if response.get("error"):
|
|
raise HostFunctionError(response["error"])
|
|
|
|
return response.get("result", "")
|
|
|
|
|
|
def task_get(task_id: str) -> Any:
|
|
"""Get returns the current state of a task including its status,
|
|
message, and attempt count.
|
|
|
|
Args:
|
|
task_id: str parameter.
|
|
|
|
Returns:
|
|
Any: The result value.
|
|
|
|
Raises:
|
|
HostFunctionError: If the host function returns an error.
|
|
"""
|
|
request = {
|
|
"taskId": task_id,
|
|
}
|
|
request_bytes = json.dumps(request).encode("utf-8")
|
|
request_mem = extism.memory.alloc(request_bytes)
|
|
response_offset = _task_get(request_mem.offset)
|
|
response_mem = extism.memory.find(response_offset)
|
|
response = json.loads(extism.memory.string(response_mem))
|
|
|
|
if response.get("error"):
|
|
raise HostFunctionError(response["error"])
|
|
|
|
return response.get("result", None)
|
|
|
|
|
|
def task_cancel(task_id: str) -> None:
|
|
"""Cancel cancels a pending task. Returns error if already
|
|
running, completed, or failed.
|
|
|
|
Args:
|
|
task_id: str parameter.
|
|
|
|
Raises:
|
|
HostFunctionError: If the host function returns an error.
|
|
"""
|
|
request = {
|
|
"taskId": task_id,
|
|
}
|
|
request_bytes = json.dumps(request).encode("utf-8")
|
|
request_mem = extism.memory.alloc(request_bytes)
|
|
response_offset = _task_cancel(request_mem.offset)
|
|
response_mem = extism.memory.find(response_offset)
|
|
response = json.loads(extism.memory.string(response_mem))
|
|
|
|
if response.get("error"):
|
|
raise HostFunctionError(response["error"])
|
|
|
|
|
|
|
|
def task_clear_queue(queue_name: str) -> int:
|
|
"""ClearQueue removes all pending tasks from the named queue.
|
|
Running tasks are not affected. Returns the number of tasks removed.
|
|
|
|
Args:
|
|
queue_name: str parameter.
|
|
|
|
Returns:
|
|
int: The result value.
|
|
|
|
Raises:
|
|
HostFunctionError: If the host function returns an error.
|
|
"""
|
|
request = {
|
|
"queueName": queue_name,
|
|
}
|
|
request_bytes = json.dumps(request).encode("utf-8")
|
|
request_mem = extism.memory.alloc(request_bytes)
|
|
response_offset = _task_clearqueue(request_mem.offset)
|
|
response_mem = extism.memory.find(response_offset)
|
|
response = json.loads(extism.memory.string(response_mem))
|
|
|
|
if response.get("error"):
|
|
raise HostFunctionError(response["error"])
|
|
|
|
return response.get("result", 0)
|