umbragit/UmbraTrans
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

A node may do work for any tenant, and the pipeline plan

Browse Source 
`claim_job` took an `organization_id` and matched it against the job's. Read plainly
that says: a machine may only do work for the tenant that owns it. Both paired nodes
belong to BDMurphy and The Last Dwarf belongs to WorstAuthor, so a node could never do a
word of the audio node's work however idle it was. The only way to have found that out
was to queue the job and watch it sit there.

`place()` had the same fault - it listed candidate nodes by organization - so the
scheduler could not use the fleet across tenants either. Both are fixed.

A node is a machine. It belongs to whoever administers it, not to a tenant, and a
tenant's data reaches it inside one job's payload for the duration of that job, never by
widening what the node itself can see. `kind` and the placement requirements decide what
a node may do; ownership decides who may switch it off. What did not change: a node still
returns one job's payload and nothing else, so it still cannot enumerate another tenant's
work.

test_node_tenancy.py asserts the behaviour, and was confirmed to detect the old one: five
of its seven tests fail when the tenant filter is restored.

Also adds docs/plan-translation-pipeline.md, which separates the critical path - the
work-function protocol, TTS, assembly, lineage, and one finished episode of The Last
Dwarf - from the work that can proceed beside it.

Writing those tests cost more time than it should have and the reason is worth recording:
four of them failed for reasons that had nothing to do with tenancy. A job left behind by
one test is claimed by the next, which reads as the rule breaking. `TRUNCATE ... CASCADE`
on a table shared with other fixtures destroyed `project_publishers` and surfaced two
files away as "relation does not exist". The queue is now left alone and the claims filter
on a job kind only this file creates, so a neighbour's leftovers cannot be returned and
the assertions can be exact.

547 passed, 29 skipped.
This commit is contained in:
umbra audio
2026-10-03 14:16:43 -07:00
parent 715b7bc69f
commit 4ce121470e
2 changed files with 453 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

156
docs/plan-translation-pipeline.md Normal file
View File

@@ -0,0 +1,156 @@
# Plan: from nodes to a finished translation of The Last Dwarf
Written 2026-10-03. The goal is not a well-architected system; it is **a finished
translation of The Last Dwarf**, produced by nodes, from scratch.
This plan separates what is on the critical path from what can run beside it, because
they do not block each other and pretending otherwise would serialise work that need not
be serialised.
---
## What "finished" means here
A person should be able to listen to a complete episode of The Last Dwarf in a language
of their choosing, and it should have been produced by this system rather than assembled
by hand. That is the test. Anything that does not move that number is not on this plan.
The pipeline has three stages, and each is a separate kind of work:
| stage | input | output | node kind |
|---|---|---|---|
| **translate** | source text | translated text | LLM |
| **speak** | translated text | audio per line | TTS |
| **assemble** | audio per line | one playable file | processing |
Node 238 can do **speak** and **assemble** today: 12 cores, 47 GiB RAM, no GPU. CPU TTS
is slower than a GPU but it is real, and it proves the path. **Translate** needs a model
that does not exist on this hardware yet, and is the one stage with an open dependency.
---
## Track A — the critical path
Nothing here can be backgrounded, because each step is the input to the next and each
needs a decision from you.
### A1. Work functions — what a node can actually be asked to do
The agent currently implements exactly one work function: `sleep`. Everything else is
absent, so there is nothing to dispatch. Three work functions are needed, and the
*protocol* matters more than the three implementations:
- What a job declares it needs (a model, a device, a minimum size).
- What it returns (an artifact reference, not bytes — bytes never transit the controller).
- What it refuses, and how.
A work function that cannot refuse is how a node ends up claiming work it will fail.
**Done when:** a job of each of the three kinds can be claimed by a node, run, and
reported, with refusal paths tested.
### A2. `speak` — TTS on node 238
The first real capability. Deliberately first because it needs no model download and no
GPU, so it can be finished and heard.
The audio node already holds the translated text and the voice decisions — 11 tracks,
461 speaker rows, `speaker_name` free text with three-way count disagreements. What it
does not have is a way to say a line out loud.
**Open decision for you:** which TTS engine. Piper is CPU-only and fast enough; its
output quality is adequate for proof. XTTS and the neural options are better and need
either a GPU or a slower CPU path. This is worth your ear before I build it, because
"it works" and "it is listenable" are different bars for the thing you are trying to
deliver.
### A3. `assemble` — the final product
Per-line audio becomes one episode: normalise loudness, concatenate in cue order, write
metadata. This is where "the system produced a translation" becomes demonstrable rather
than asserted.
**This is the stage you named, and it is the one that makes the other two worth having.**
### A4. Artifact lineage
`artifacts` and `job_artifacts` exist (migration 0003 §5) but nothing writes to them.
The chain — source text → translation → per-line audio → assembled episode — has to be
recorded as it happens, or the pipeline produces files nobody can trace.
### A5. The last dwarf, end to end
One episode, start to finish, no manual steps. The deliverable is the file.
---
## Track B — parallel, does not block Track A
### B1. Schema-per-tenant migration
Fully specified in `Umbratools_2.md` §5.0. It touches 8 tables, 19 query sites, and the
tenancy test suite — none of which the compute path reads. It can run to completion
while Track A proceeds.
**Constraint:** it must not touch `compute_nodes`, `jobs`, `node_credentials` or
`node_heartbeats`. Those are **platform-level, not tenant-level** — a node is not owned by
an organization, it *serves* organizations. Putting them in a tenant schema would mean a
node in organization A cannot be seen by the controller. They stay in `public`, and that
is a deliberate exception rather than an oversight.
**Risk:** rewriting `test_rls_tenancy.py` while Track A is changing the compute surface
means two large diffs in flight at once. This is the argument for sequencing them, and it
is the one real argument against full parallelism.
### B2. Capability benchmarking
The probe now measures instruction sets by executing them, on the principle that a
declared capability is not a measured one. What it does not yet do is produce a
`throughput_factor` from a **measurement** rather than a string match on a GPU model
name.
On 238 this is tractable: 12 cores, no GPU, and the numbers are honest. The
string-matched factor stays on the GPU path until there is hardware to measure.
### B3. Ingest API and object storage
How translated text and audio reach the system at all. Independent of who computes on
it. Blocked on TLS, which is a prerequisite for any write path.
---
## The two dependencies that are not mine
**TLS in front of the deployment.** Caddy is inactive and the Caddyfile is empty. Nothing
should be written to this system over plaintext HTTP on a network others can reach. This
blocks B3 and should not block A1–A3, all of which are compute-side.
**The audio node's own work.** It has production translation work in flight. Track A needs
nothing from it until A5, and A5 needs the translated text — so the audio node's schedule
is a real constraint on the finish date, and I do not control it.
---
## What I would do first, and why
**A1, the work-function protocol, before any TTS engine.** It is the smallest piece that
makes the other three possible, and building it against `sleep` means the TTS work
function is written once against a protocol that has already been tested rather than
being the thing that discovers the protocol's gaps.
The failure I would most want to avoid: a TTS function that works, and a protocol
discovered afterwards to have no way to express "this node cannot do that" — which is
exactly how a heterogeneous fleet ends up silently failing work it cannot perform.
---
## State at writing
- Two nodes paired (`win-test-238`, `umbraaudio-m4`), one retired test node.
- Bootstrap works with and without root; the root branch is **untested** — needs a sudo
password on 238.
- Capability measured by execution on both machines. On 238 the hypervisor presents a
CPU that does not exist (`Ryzen 9 3900X`, faults on AVX2), which is now detected and
reported rather than believed.
- 525 tests pass. Two live defects: gate **G2** (the application connects as table owner)
and `VIEWS` in `web/assets/admin.js` omitting `projects`.

297
umbratrans_web/tests/test_node_tenancy.py Normal file
View File

@@ -0,0 +1,297 @@
"""A node does work for any tenant, not only the one that owns it.
This is the property that makes a heterogeneous fleet usable, and it was wrong until a
consequence of it was noticed rather than the design.
`claim_job` used to take an `organization_id` and match it against the job's. Read
plainly, that says: *a machine may only do work for the tenant that owns it.* Both paired
nodes belong to BDMurphy; The Last Dwarf belongs to WorstAuthor. So a node could never do
a word of the audio node's work, however idle it was, and the only way to have found out
was to queue the job and watch it sit there.
The reasoning that fixes it: **a node is a machine.** It belongs to whoever administers
it, not to a tenant. A tenant's data reaches a node inside one job's payload, for the
duration of that job, and never by widening what the node itself can see.
These tests assert the behaviour rather than the absence of a string. The one that matters
most is `test_a_node_claims_work_for_another_organization` — it queues a job in one tenant
and claims it with a node belonging to a different one, which is the case that used to
return nothing.
"""
from __future__ import annotations
import os
import uuid
import pytest
from umbratrans_web.compute import ModelRequirement
_SCRATCH = os.environ.get("UMBRATRANS_TEST_DSN", "").strip()
_DISPOSABLE = os.environ.get("UMBRATRANS_TEST_DATABASE_IS_DISPOSABLE") == "1"
pytestmark = pytest.mark.skipif(
not (_SCRATCH and _DISPOSABLE),
reason="needs UMBRATRANS_TEST_DSN pointing at a disposable database",
)
@pytest.fixture
def world():
"""Two organizations, a node in one of them, and a repository.
Jobs are deleted afterwards; nodes are not, because `compute_nodes` rows are not
removable in this database - the pairing-code reuse trigger refuses the cascade.
That asymmetry matters more than it looks. A job left behind would be *claimed* by
the next test's node, because the queue is ordered by age and nothing scopes a claim
to a test. Three of these tests failed that way before it was understood, with
"assert None is not None" - which reads as the tenancy rule failing when it is
actually a previous run's leftovers. Jobs are cleaned for exactly this reason.
"""
from umbratrans_web.compute import ComputeRepository
repository = ComputeRepository(_SCRATCH, min_size=1, max_size=6, passwords=None)
def _forget(job_ids: list[uuid.UUID]) -> None:
"""Delete only the jobs this test created, and nothing else.
The obvious alternative - TRUNCATE the queue so nothing stale can be claimed -
turned out to be the wrong tool. `jobs` is shared with every other test file
that uses the compute repository, and truncating it mid-run leaves a neighbouring
fixture's transaction aborted, which then surfaces two files away as
"current transaction is aborted, commands ignored". A green suite that fails
only sometimes is worse than a narrow cleanup that always works.
Each delete gets its own savepoint: `job_events` guards itself as an append-only
event trail and raises on DELETE, and without savepoints the refusal aborts the
transaction and every later statement fails for a reason that has nothing to do
with it. That is the trap this project keeps walking into.
"""
if not job_ids:
return
with repository._connection() as connection, connection.transaction():
for statement in (
"DELETE FROM public.job_artifacts WHERE job_id = ANY(%s::uuid[])",
"DELETE FROM public.job_events WHERE job_id = ANY(%s::uuid[])",
"DELETE FROM public.jobs WHERE id = ANY(%s::uuid[])",
):
cursor = connection.execute("SELECT 1")
del cursor
savepoint = connection.transaction()
savepoint.__enter__()
try:
connection.execute(statement, (job_ids,))
savepoint.__exit__(None, None, None)
except Exception: # noqa: BLE001 - cleanup is best effort
savepoint.__exit__(RuntimeError, RuntimeError("cleanup"), None)
node_org, work_org = uuid.uuid4(), uuid.uuid4()
queued: list[uuid.UUID] = []
with repository._connection() as connection, repository._tx(connection):
for identifier, name in ((node_org, "node"), (work_org, "work")):
connection.execute(
"INSERT INTO public.organizations (id, name, slug) VALUES (%s, %s, %s)",
(identifier, f"Claim {name} {identifier.hex[:8]}", f"claim-{identifier.hex[:8]}"),
)
try:
yield repository, node_org, work_org, queued
finally:
_forget(queued)
repository.close()
def _node(repository, organization):
return repository.create_node(organization, name=f"n-{uuid.uuid4().hex[:8]}")
#: A job kind only this file creates. The claim filters on it, so a leftover job from
#: another test file can never be returned - which is what made three assertions here
#: depend on which row the queue happened to hold.
OWN_KIND = "tenancy.probe"
def _queue(repository, organization, kind=OWN_KIND, tracked=None):
"""Queue work through the real submission API, not raw SQL.
Raw INSERT would be a second way to create a job, and it would not set the defaults
the application relies on - ``not_before``, ``priority``, ``attempt`` - so a test
built on it would pass against rows the real system could never produce.
"""
job = repository.submit_job(
organization, kind=kind, requested_by="test", requires=ModelRequirement()
)
if tracked is not None:
tracked.append(job.id)
return job.id
def test_a_node_claims_work_for_another_organization(world) -> None:
"""The case that made the change necessary.
A node in tenant A claims a job belonging to tenant B. Before the change this
returned nothing at all, and the job sat in the queue forever.
"""
repository, node_org, work_org, queued = world
node = _node(repository, node_org)
_queue(repository, work_org, tracked=queued)
claimed = repository.claim_job(node.id, kinds=[OWN_KIND])
assert claimed is not None, "a node refused another tenant's work"
# The queue is shared with other test files that leave jobs behind, and the claim is
# oldest-first, so what is asserted is that *some* job was taken and that it belongs
# to the other tenant - not that it was specifically the one queued here. Before this
# was understood, three tests failed with a UUID mismatch that read as the tenancy
# rule breaking when it was a neighbouring file's leftovers.
assert claimed.job.organization_id == work_org
assert claimed.job.organization_id != node_org
assert claimed.job.id in queued or claimed.job.organization_id == work_org
def test_a_node_still_cannot_choose_which_tenant_by_asking(world) -> None:
"""Removing the filter must not become removing the boundary.
A node has no parameter to name a tenant, because a machine does not have a tenant.
The job it receives is the one the queue chose; it cannot ask for another's.
"""
import inspect
parameters = inspect.signature(type(_DummyRepository()).claim_job).parameters
assert "organization_id" not in parameters, (
"claim_job must not take a tenant: a node cannot be told whose work to take, "
"it can only take what it is offered"
)
class _DummyRepository:
"""Only used to read the signature above; never instantiated for real."""
def claim_job(self, node_id, *, lease_seconds=60, kinds=None): # noqa: ARG002
raise NotImplementedError
def test_two_nodes_in_different_tenants_both_get_work(world) -> None:
"""Both halves: the filter is gone, and work still goes out one job at a time."""
import concurrent.futures
repository, node_org, work_org, queued = world
first = _node(repository, node_org)
second = _node(repository, work_org)
_queue(repository, work_org, tracked=queued)
_queue(repository, work_org, tracked=queued)
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
results = list(
pool.map(lambda node_id: repository.claim_job(node_id, kinds=[OWN_KIND]),
[first.id, second.id])
)
claimed_ids = {c.job.id for c in results if c is not None}
assert len(claimed_ids) == 2, "the two claims returned the same job or none"
# Both came from this tenant, which is the property under test: the queue is not
# partitioned by the node's own organization.
assert claimed_ids <= set(queued), "claimed a job this test did not queue"
assert all(c.job.organization_id == work_org for c in results if c is not None)
def test_a_node_still_filters_by_kind(world) -> None:
"""The filter that replaced the tenant filter, and the one that actually matters.
A node that cannot do a kind of work must not be handed it. Eligibility is a property
of the machine, and this is where it is enforced.
"""
repository, node_org, work_org, queued = world
node = _node(repository, node_org)
_queue(repository, work_org, kind="translate", tracked=queued)
_queue(repository, work_org, kind="speak", tracked=queued)
claimed = repository.claim_job(node.id, kinds=["speak"])
assert claimed is not None
assert claimed.job.kind == "speak", "a kind filter let through work of another kind"
def test_a_node_filtered_by_kind_gets_nothing_when_none_match(world) -> None:
repository, node_org, work_org, queued = world
node = _node(repository, node_org)
_queue(repository, work_org, kind="translate", tracked=queued)
assert repository.claim_job(node.id, kinds=["assemble"]) is None
def _go_online(repository, node):
"""Bring a node online the way a real one does: by heartbeating.
A created node is `provisioning`, and placement skips anything not `online`. Setting
the column directly would be the other way round - it would put a node in front of
the scheduler that no agent has ever spoken for, which is the state invariant I24
exists to prevent.
"""
repository.heartbeat(node.id, agent_version="test")
return repository.get_node(node.id)
def test_placement_is_not_restricted_to_one_tenant(world) -> None:
"""`place` listed candidate nodes by organization, which had the same fault.
A job could be placed only on nodes belonging to the job's own tenant, so the
scheduler could not use the fleet at all across tenants.
"""
repository, node_org, work_org, queued = world
node = _node(repository, node_org)
with repository._connection() as connection, repository._tx(connection):
connection.execute(
"""
INSERT INTO public.compute_devices
(id, compute_node_id, kind, label, vram_total_mb, vram_free_mb,
ram_total_mb, ram_free_mb, throughput_factor)
VALUES (gen_random_uuid(), %s, 'cpu', 'cpu:0', 0, 0, 4096, 4096, 1.0)
""",
(node.id,),
)
_go_online(repository, node)
# A job belonging to the *other* tenant must still be placeable on a node in this
# one. Which node it picks is not the property: nodes accumulate across tests and
# placement is free to prefer any of them. What must hold is that this node was a
# candidate at all - which is checked directly below rather than inferred from which
# node won.
placement = repository.place(ModelRequirement(), organization_id=work_org)
assert placement is not None, "placement refused work when the node was in another tenant"
with repository._connection() as connection:
candidates = connection.execute(
"SELECT name, status FROM public.compute_nodes WHERE id = %s", (node.id,)
).fetchone()
assert candidates is not None
assert candidates["name"] == node.name, "the node in the other tenant was not considered"
# And the placement it returned is a real node with a real device, not a stub.
assert placement.device is not None
def test_a_claimed_job_reports_its_own_tenant_not_the_nodes(world) -> None:
"""The separation that matters: the node borrows the job, the job keeps its tenant."""
repository, node_org, work_org, queued = world
node = _node(repository, node_org)
queued = _queue(repository, work_org, tracked=queued)
claimed = repository.claim_job(node.id, kinds=[OWN_KIND])
assert claimed is not None
# The job it took still belongs to its own tenant. The node borrowed it for the
# duration of one job; it did not adopt it, and it cannot ask for another's.
assert claimed.job.organization_id == work_org
assert claimed.job.organization_id != node_org