Skip to content

lablink_allocator_service.operations

Background worker for on-demand OpenTofu apply/destroy operations.

Runs each operation on its own daemon thread, off the Flask request thread, so a slow tofu apply/destroy doesn't hold an HTTP connection open long enough to hit Cloudflare's edge timeout.

Submits apply/destroy jobs to run on a background thread.

Parameters:

Name Type Description Default
database 'OperationsDatabase'

OperationsDatabase instance for operation-status persistence.

 required
Source code in packages/allocator/src/lablink_allocator_service/operations.py 
26
27
def __init__(self, database: "OperationsDatabase"):
    self.database = database

Mark any operation left queued/running from a prior process as interrupted. Call once at allocator startup, before any new operation can be submitted, so a crash/restart mid-job is surfaced instead of hanging forever or being silently forgotten.

Source code in packages/allocator/src/lablink_allocator_service/operations.py 
29
30
31
32
33
34
35
36
37
38
39
40
def start(self) -> None:
    """Mark any operation left queued/running from a prior process as
    interrupted. Call once at allocator startup, before any new
    operation can be submitted, so a crash/restart mid-job is surfaced
    instead of hanging forever or being silently forgotten."""
    count = self.database.sweep_interrupted_operations()
    if count:
        logger.warning(
            "Marked %d operation(s) interrupted "
            "(allocator restarted mid-job)",
            count,
        )

Queue a new operation and run it on a background thread.

Returns immediately after the operation row is created — does not wait for fn to complete.

Raises:

Type Description
OperationInProgress

propagated from database.create_operation if another operation is already queued/running.
Source code in packages/allocator/src/lablink_allocator_service/operations.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def submit(
    self,
    op_type: str,
    fn: Callable[..., str],
    params: Optional[str] = None,
    created_by: Optional[str] = None,
) -> int:
    """Queue a new operation and run it on a background thread.

    Returns immediately after the operation row is created — does not
    wait for `fn` to complete.

    Raises:
        OperationInProgress: propagated from database.create_operation
            if another operation is already queued/running.
    """
    operation_id = self.database.create_operation(
        op_type=op_type, params=params, created_by=created_by
    )
    Thread(
        target=self._run, args=(operation_id, fn), daemon=True
    ).start()
    return operation_id