DELETE /artifacts/{digest}: superuser only, 409 while referenced
The keep-set the hourly sweep uses is now referenced_artifacts, shared by the sweep and the route. The volatile ring is not touched.
This commit is contained in:
@@ -10,17 +10,23 @@ import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, Request
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
|
||||
from fastapi.responses import FileResponse
|
||||
from jwt.exceptions import InvalidTokenError
|
||||
from pydantic import BaseModel
|
||||
from sqlmodel import Session
|
||||
from starlette.concurrency import run_in_threadpool
|
||||
|
||||
from fluksio.api.deps import user_from_token, worker_claims
|
||||
from fluksio.api.deps import (
|
||||
FlowControllerDep,
|
||||
get_current_active_superuser,
|
||||
user_from_token,
|
||||
worker_claims,
|
||||
)
|
||||
from fluksio.core.config import settings
|
||||
from fluksio.core.db import engine
|
||||
from fluksio.flow.artifacts import ArtifactStore
|
||||
from fluksio.flow.runs import referenced_artifacts
|
||||
|
||||
#: What may be echoed back as a response content type. The caller holds the
|
||||
#: reference and passes its media type, so this guards a header rather than
|
||||
@@ -153,3 +159,33 @@ def get_artifact(
|
||||
headers["Content-Security-Policy"] = "sandbox"
|
||||
headers["X-Content-Type-Options"] = "nosniff"
|
||||
return FileResponse(path, media_type=media_type, headers=headers)
|
||||
|
||||
|
||||
@router.delete(
|
||||
"/{digest}",
|
||||
status_code=204,
|
||||
dependencies=[Depends(get_current_active_superuser)],
|
||||
)
|
||||
def delete_artifact(
|
||||
digest: str, request: Request, controller: FlowControllerDep
|
||||
) -> Response:
|
||||
"""Remove one artifact's bytes from the store.
|
||||
|
||||
A superuser's call, because the store records no uploader and identical
|
||||
content from two callers is one file, so nobody else can be said to own
|
||||
it. Refused while anything still names the digest, the same references the
|
||||
hourly sweep keeps, so a reference that exists never answers 404. The
|
||||
volatile ring is left alone: its frames fall out on their own.
|
||||
"""
|
||||
store = _store(request)
|
||||
if digest in referenced_artifacts(controller.state, controller.store):
|
||||
raise HTTPException(
|
||||
status_code=409,
|
||||
detail=(
|
||||
"The artifact is still referenced by a run, a current message or "
|
||||
"a flow input's initial value"
|
||||
),
|
||||
)
|
||||
if not store.delete(digest):
|
||||
raise HTTPException(status_code=404, detail="No such artifact")
|
||||
return Response(status_code=204)
|
||||
|
||||
@@ -160,6 +160,19 @@ class ArtifactStore:
|
||||
self.put_file(source)
|
||||
return True
|
||||
|
||||
def delete(self, digest: str) -> bool:
|
||||
"""Remove one artifact's bytes. Returns whether this store had them.
|
||||
|
||||
The ring is not looked in: its frames fall out on their own.
|
||||
"""
|
||||
if not valid_digest(digest):
|
||||
return False
|
||||
try:
|
||||
self._path(digest).unlink()
|
||||
except FileNotFoundError:
|
||||
return False
|
||||
return True
|
||||
|
||||
def read(self, digest: str) -> Iterator[bytes]:
|
||||
target = self.path(digest)
|
||||
if target is None:
|
||||
|
||||
@@ -382,13 +382,12 @@ def sweep_artifacts(
|
||||
) -> int:
|
||||
"""Remove artifact bytes nothing refers to any more. Returns how many went.
|
||||
|
||||
Three things refer to an artifact: a run that recorded it, a message
|
||||
currently holding it, and a flow input's ``initial``, published or draft,
|
||||
which a flow that has never started holds nowhere else. The second is what
|
||||
makes a stream of media affordable — a camera publishing a frame a second
|
||||
replaces the reference each time, so yesterday's frames are unreferenced by
|
||||
definition and the store does not grow without bound. What a node *returns*
|
||||
is recorded against its run and kept, what it emits along the way is not.
|
||||
What still refers to one is `referenced_artifacts`. A message holding one is
|
||||
what makes a stream of media affordable — a camera publishing a frame a
|
||||
second replaces the reference each time, so yesterday's frames are
|
||||
unreferenced by definition and the store does not grow without bound. What
|
||||
a node *returns* is recorded against its run and kept, what it emits along
|
||||
the way is not.
|
||||
|
||||
Skipped entirely while anything is running: a node that stores a checkpoint
|
||||
an hour before it returns has neither a row nor a message naming it yet,
|
||||
@@ -401,6 +400,23 @@ def sweep_artifacts(
|
||||
if active is not None:
|
||||
logger.debug("Artifact sweep skipped: run %s is in flight", active)
|
||||
return 0
|
||||
|
||||
removed = store.collect(referenced_artifacts(state, flows), grace_s)
|
||||
if removed:
|
||||
logger.info("Artifact sweep removed %d unreferenced artifacts", removed)
|
||||
return removed
|
||||
|
||||
|
||||
def referenced_artifacts(
|
||||
state: StateBackend, flows: FlowStore | None = None
|
||||
) -> set[str]:
|
||||
"""Every artifact digest something still refers to.
|
||||
|
||||
Three things do: a run that recorded it, a message currently holding it,
|
||||
and a flow input's ``initial``, published or draft, which a flow that has
|
||||
never started holds nowhere else.
|
||||
"""
|
||||
with Session(db_engine) as session:
|
||||
keep = set(session.exec(select(col(RunArtifact.digest)).distinct()).all())
|
||||
|
||||
if flows is not None:
|
||||
@@ -417,11 +433,7 @@ def sweep_artifacts(
|
||||
for start in range(0, len(keys), _SWEEP_BATCH):
|
||||
for value in state.get_multi(keys[start : start + _SWEEP_BATCH]).values():
|
||||
_references_in(value, keep)
|
||||
|
||||
removed = store.collect(keep, grace_s)
|
||||
if removed:
|
||||
logger.info("Artifact sweep removed %d unreferenced artifacts", removed)
|
||||
return removed
|
||||
return keep
|
||||
|
||||
|
||||
def seed_values(
|
||||
|
||||
@@ -230,3 +230,33 @@ def test_a_type_that_can_run_script_is_served_sandboxed(
|
||||
assert headers["content-security-policy"] == "sandbox"
|
||||
assert headers["x-content-type-options"] == "nosniff"
|
||||
assert "content-security-policy" not in served(image, "image/png")
|
||||
|
||||
|
||||
def test_a_superuser_deletes_an_artifact_nothing_references(
|
||||
client: TestClient,
|
||||
superuser_token_headers: dict[str, str],
|
||||
normal_user_token_headers: dict[str, str],
|
||||
) -> None:
|
||||
url = f"{settings.API_V1_STR}/artifacts"
|
||||
digest = client.put(
|
||||
f"{url}?name=x.bin", headers=superuser_token_headers, content=b"delete me"
|
||||
).json()["digest"]
|
||||
|
||||
refused = client.delete(f"{url}/{digest}", headers=normal_user_token_headers)
|
||||
assert refused.status_code == 403
|
||||
|
||||
run_id = _run_row("ok", digest)
|
||||
try:
|
||||
held = client.delete(f"{url}/{digest}", headers=superuser_token_headers)
|
||||
assert held.status_code == 409
|
||||
finally:
|
||||
_forget(run_id)
|
||||
|
||||
gone = client.delete(f"{url}/{digest}", headers=superuser_token_headers)
|
||||
assert gone.status_code == 204
|
||||
assert (
|
||||
client.get(f"{url}/{digest}", headers=superuser_token_headers).status_code
|
||||
== 404
|
||||
)
|
||||
again = client.delete(f"{url}/{digest}", headers=superuser_token_headers)
|
||||
assert again.status_code == 404
|
||||
|
||||
@@ -172,6 +172,11 @@ curl "$FLUKSIO/artifacts/<digest>" \
|
||||
The upload returns a reference with the content digest. Identical bytes are
|
||||
stored once. Node code uses `save_artifact` and `load_artifact` instead.
|
||||
|
||||
A superuser can remove stored bytes with `DELETE /artifacts/<digest>`. While
|
||||
a run, a current message or a flow input's initial value still references the
|
||||
digest, the call returns `409`. The hourly sweep removes unreferenced bytes
|
||||
on its own, so this is for reclaiming space sooner.
|
||||
|
||||
## Dashboards and panels
|
||||
|
||||
| Method | Path | Action |
|
||||
|
||||
File diff suppressed because one or more lines are too long
@@ -1620,6 +1620,12 @@ export type ArtifactsGetArtifactData = {
|
||||
|
||||
export type ArtifactsGetArtifactResponse = (unknown);
|
||||
|
||||
export type ArtifactsDeleteArtifactData = {
|
||||
digest: string;
|
||||
};
|
||||
|
||||
export type ArtifactsDeleteArtifactResponse = (void);
|
||||
|
||||
export type CloudReadStatusResponse = ({
|
||||
[key: string]: unknown;
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user