Merge pull request 'v0.32.0: per-recording delete and nav link, cloud-served note, tracker hardening, bounded LLM reads (#37 #51 #52 #53)' (#58) from sprint/v032 into main
CI / Deploy via GitOps (push) Successful in 5s
CI / Lint / Test / Vet (push) Successful in 34s
CI / Build & Import (push) Successful in 11s

This commit was merged in pull request #58.
This commit is contained in:
2026-10-03 22:31:26 +00:00
27 changed files with 1696 additions and 718 deletions
+8 -2
View File
@@ -1551,8 +1551,14 @@ deferred. The scope cut — chat first, no rich analysis, no video — is what k
disabled by default, a failed iguana job fails honestly and the user retries. *Blocks:* disabled by default, a failed iguana job fails honestly and the user retries. *Blocks:*
the production default only; the adapter ships either way. the production default only; the adapter ships either way.
- ~~Verify Tapir's ingress `proxy-body-size` against the 200 MB cap (`infra`).~~ **Resolved 2026-10-03:** real unauthenticated bodies of 2, 30, 120 and 210 MiB all pass NPM (TLS, piguard) and Traefik to tapir (a 302 to login); there is no body limit in that range. Uploads were enabled in production the same day, local engines only (infra#520). - ~~Verify Tapir's ingress `proxy-body-size` against the 200 MB cap (`infra`).~~ **Resolved 2026-10-03:** real unauthenticated bodies of 2, 30, 120 and 210 MiB all pass NPM (TLS, piguard) and Traefik to tapir (a 302 to login); there is no body limit in that range. Uploads were enabled in production the same day, local engines only (infra#520).
- Per-transcript deletion is not built; account deletion cascades. A private-content store - ~~Per-transcript deletion is not built; account deletion cascades.~~ **Built 2026-10-03
without a delete button is a gap worth a ticket, not a silent omission. (#37):** `POST /uploads/{id}/delete`, owner-scoped (a non-owner gets 404 by the same RLS
mechanism), confirm-then-delete on the recording's page. A recording that is **queued or
transcribing is refused (409) until it ends**, because the worker would otherwise write to a
row that is gone; a job whose service was lost stays undeletable until the next start's sweep
marks it failed. The refusal reflects the row's state when the delete was attempted: a
recording that finishes in the same instant still gets a 409 once, and a retry deletes it.
The home list now links to `/uploads` when uploads are mounted.
**Reversibility.** Additive: one table (down-migration drops it), one adapter, one upload **Reversibility.** Additive: one table (down-migration drops it), one adapter, one upload
surface, one image change. Removing the upload route removes the feature; `transcripts`, `videos` surface, one image change. Removing the upload route removes the feature; `transcripts`, `videos`
+1 -1
View File
@@ -44,7 +44,7 @@ func (f *fakeUploadsStore) SaveProgress(context.Context, string, string, []domai
return nil return nil
} }
func (f *fakeUploadsStore) Heartbeat(context.Context, string, string) error { return nil } func (f *fakeUploadsStore) Heartbeat(context.Context, string, string, time.Time) error { return nil }
func (f *fakeUploadsStore) FailInterruptedUploads(context.Context) (int, error) { func (f *fakeUploadsStore) FailInterruptedUploads(context.Context) (int, error) {
f.sweeps++ f.sweeps++
+2 -1
View File
@@ -14,7 +14,8 @@ it** — endpoints and aliases drift, and this file is a snapshot (2026-06-06),
not LiteLLM. Use `30401` for off-cluster LiteLLM. `confirm` resolved. not LiteLLM. Use `30401` for off-cluster LiteLLM. `confirm` resolved.
- **Speech-to-text (ADR-033):** the same gateway serves `POST /v1/audio/transcriptions` for - **Speech-to-text (ADR-033):** the same gateway serves `POST /v1/audio/transcriptions` for
`iguana/kb-whisper` (primary, local) and `berget/kb-whisper` (fallback, EU cloud — private audio `iguana/kb-whisper` (primary, local) and `berget/kb-whisper` (fallback, EU cloud — private audio
leaves the homelab if it is used). **Call it via the NodePort / in-cluster address above, NOT leaves the homelab if it is used: when the primary fails **or returns an empty transcript**,
e.g. a silent or music-only file; the transcript page then says so, #52). **Call it via the NodePort / in-cluster address above, NOT
`https://llm-api.d-ma.be`:** that public ingress (openresty) returns `413` for bodies over `https://llm-api.d-ma.be`:** that public ingress (openresty) returns `413` for bodies over
roughly 8–10 MB (~5 minutes of 16 kHz WAV), while the NodePort took 174 MB (90 minutes) in one roughly 8–10 MB (~5 minutes of 16 kHz WAV), while the NodePort took 174 MB (90 minutes) in one
request (measured 2026-10-01, `docs/research/audio-length-ceiling-2026-10-01.md`). Send multipart request (measured 2026-10-01, `docs/research/audio-length-ceiling-2026-10-01.md`). Send multipart
+18
View File
@@ -240,6 +240,24 @@ Feature: Upload an audio recording, get a private Swedish transcript, discuss it
Then it is transcribed anyway and its audio step says how long the audio really is Then it is transcribed anyway and its audio step says how long the audio really is
And a complete recording whose header over-estimated its length is not refused And a complete recording whose header over-estimated its length is not refused
Scenario: A transcript made by the cloud fallback says the audio left the homelab
When the transcript was produced by an endpoint that is not local
Then its page says in plain words that the audio left the homelab and names the provider
And a locally transcribed recording carries no such note
Scenario: I can delete a finished recording
When I confirm deleting a done or failed recording
Then the recording and its transcript are gone and I am back on my list
And another user's recording is a 404 and is untouched
Scenario: A recording still being transcribed cannot be deleted
When I try to delete a recording that is queued or transcribing
Then I am told it is still being transcribed and nothing is removed
Scenario: The home list links to my recordings when uploads are on
Given audio upload is enabled
Then the home list links to the recordings page
Scenario: A full queue is a retry, not an error Scenario: A full queue is a retry, not an error
When too many recordings are already waiting When too many recordings are already waiting
Then the upload is answered as busy, its row is failed honestly and its audio is deleted Then the upload is answered as busy, its row is failed honestly and its audio is deleted
+156
View File
@@ -0,0 +1,156 @@
package llm
// Observes how much of a gateway response the client will read. The body is read before
// the status is looked at, so a misbehaving gateway or proxy that streams forever was
// bounded only by the client's overall timeout (#53).
import (
"context"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"time"
)
// endlessGateway answers with the given status and streams filler until the client hangs
// up; wrote reports how many bytes it managed to send.
func endlessGateway(t *testing.T, status int, wrote *atomic.Int64) *httptest.Server {
t.Helper()
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(status)
chunk := []byte(strings.Repeat("a", 4096))
for {
n, err := w.Write(chunk)
wrote.Add(int64(n))
if err != nil {
return
}
if f, ok := w.(http.Flusher); ok {
f.Flush()
}
}
}))
t.Cleanup(srv.Close)
return srv
}
func TestClient_ASuccessBodyLargerThanTheCapIsAnErrorAndTheClientStopsReading(t *testing.T) {
var wrote atomic.Int64
srv := endlessGateway(t, http.StatusOK, &wrote)
c := New(srv.URL, "", "m", 30*time.Second, WithMaxResponseBytes(8192))
start := time.Now()
_, err := c.Complete(context.Background(), "s", "u")
if err == nil || !strings.Contains(err.Error(), "exceeds") {
t.Fatalf("want a response-too-large error, got %v", err)
}
if time.Since(start) > 5*time.Second {
t.Fatalf("the client kept reading: took %s", time.Since(start))
}
// Hanging up lets the server's writes fail; allow what the transport had already buffered.
time.Sleep(100 * time.Millisecond)
if got := wrote.Load(); got > 8192+1<<20 {
t.Fatalf("server managed to send %d bytes after a cap of 8192", got)
}
}
func TestClient_AFailingBodyIsReadOnlyUpToTheErrorBound(t *testing.T) {
var wrote atomic.Int64
srv := endlessGateway(t, http.StatusBadGateway, &wrote)
c := New(srv.URL, "", "m", 30*time.Second)
start := time.Now()
_, err := c.Complete(context.Background(), "s", "u")
he, ok := err.(*HTTPError)
if !ok {
t.Fatalf("want *HTTPError, got %T: %v", err, err)
}
if he.Code != http.StatusBadGateway {
t.Fatalf("code %d", he.Code)
}
if len(he.Body) > maxErrorBody+1 {
t.Fatalf("kept %d bytes of an error body, bound is %d", len(he.Body), maxErrorBody)
}
if time.Since(start) > 5*time.Second {
t.Fatalf("the client kept reading an endless error body: took %s", time.Since(start))
}
}
// sizedGateway answers 200 with a valid completion padded with trailing whitespace to
// exactly size bytes, so the boundary of the cap can be probed to the byte.
func sizedGateway(t *testing.T, size int) *httptest.Server {
t.Helper()
const base = `{"choices":[{"message":{"content":"ok"}}]}`
payload := base + strings.Repeat(" ", size-len(base))
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(payload))
}))
t.Cleanup(srv.Close)
return srv
}
func TestClient_ABodyOfExactlyTheCapIsReadAndOneByteMoreIsRefused(t *testing.T) {
const capBytes = 4096
for _, tt := range []struct {
name string
size int
wantErr bool
}{
{"one under", capBytes - 1, false},
{"exactly the cap", capBytes, false},
{"one over", capBytes + 1, true},
} {
t.Run(tt.name, func(t *testing.T) {
c := New(sizedGateway(t, tt.size).URL, "", "m", 5*time.Second, WithMaxResponseBytes(capBytes))
got, err := c.Complete(context.Background(), "s", "u")
if tt.wantErr {
if err == nil || !strings.Contains(err.Error(), "exceeds") {
t.Fatalf("want a response-too-large error, got %q, %v", got, err)
}
return
}
if err != nil || got != "ok" {
t.Fatalf("want ok, got %q, %v", got, err)
}
})
}
}
// New is what production calls: no option. The default is therefore the live bound.
func TestClient_TheDefaultCapIsEightMiBWhenNoOptionIsGiven(t *testing.T) {
for _, tt := range []struct {
name string
size int
wantErr bool
}{
{"exactly 8 MiB", 8 << 20, false},
{"8 MiB plus one", 8<<20 + 1, true},
} {
t.Run(tt.name, func(t *testing.T) {
c := New(sizedGateway(t, tt.size).URL, "", "m", 10*time.Second)
_, err := c.Complete(context.Background(), "s", "u")
if (err != nil) != tt.wantErr {
t.Fatalf("size %d: err = %v, wantErr %v", tt.size, err, tt.wantErr)
}
})
}
}
func TestClient_ANonPositiveCapOptionKeepsTheDefault(t *testing.T) {
for _, n := range []int64{0, -1} {
c := New(sizedGateway(t, 1024).URL, "", "m", 5*time.Second, WithMaxResponseBytes(n))
if got, err := c.Complete(context.Background(), "s", "u"); err != nil || got != "ok" {
t.Fatalf("WithMaxResponseBytes(%d): want ok from the default cap, got %q, %v", n, got, err)
}
}
}
+25 -3
View File
@@ -37,6 +37,21 @@ type Client struct {
temp float64 temp float64
httpClient *http.Client httpClient *http.Client
usageHook func(model string, prompt, completion int) usageHook func(model string, prompt, completion int)
maxBody int64
}
// defaultMaxResponseBytes bounds a success body. A completion is a few KB of text; this
// leaves room for a long one and none for a gateway that streams forever (#53).
const defaultMaxResponseBytes = 8 << 20
// WithMaxResponseBytes overrides the bound on a success response body. A non-positive n is
// ignored (keeps the default).
func WithMaxResponseBytes(n int64) Option {
return func(c *Client) {
if n > 0 {
c.maxBody = n
}
}
} }
// Option configures a Client at construction. Variadic so the existing 4-arg // Option configures a Client at construction. Variadic so the existing 4-arg
@@ -101,6 +116,7 @@ func New(baseURL, apiKey, model string, timeout time.Duration, opts ...Option) *
maxTokens: defaultMaxTokens, maxTokens: defaultMaxTokens,
temp: defaultTemperature, temp: defaultTemperature,
httpClient: &http.Client{Timeout: timeout}, httpClient: &http.Client{Timeout: timeout},
maxBody: defaultMaxResponseBytes,
} }
for _, opt := range opts { for _, opt := range opts {
opt(c) opt(c)
@@ -184,12 +200,18 @@ func (c *Client) Complete(ctx context.Context, system, user string) (string, err
} }
defer resp.Body.Close() //nolint:errcheck defer resp.Body.Close() //nolint:errcheck
out, err := io.ReadAll(resp.Body) if resp.StatusCode != http.StatusOK {
// Only the bounded head is ever used (HTTPError.Error truncates to it), so read no more.
// A read error here only shortens a diagnostic; the status code below is what callers act on.
head, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBody+1))
return "", &HTTPError{Code: resp.StatusCode, Body: string(head)}
}
out, err := io.ReadAll(io.LimitReader(resp.Body, c.maxBody+1))
if err != nil { if err != nil {
return "", fmt.Errorf("read response: %w", err) return "", fmt.Errorf("read response: %w", err)
} }
if resp.StatusCode != http.StatusOK { if int64(len(out)) > c.maxBody {
return "", &HTTPError{Code: resp.StatusCode, Body: string(out)} return "", fmt.Errorf("LLM response exceeds %d bytes", c.maxBody)
} }
var cr chatResponse var cr chatResponse
@@ -17,6 +17,10 @@ import (
// nothing about whether someone else's upload exists (ADR-033). // nothing about whether someone else's upload exists (ADR-033).
var ErrUploadNotFound = errors.New("store: uploaded transcript not found") var ErrUploadNotFound = errors.New("store: uploaded transcript not found")
// ErrUploadInProgress is returned by DeleteUpload for an upload that is still queued or
// transcribing: the worker would write its result to a row that no longer exists.
var ErrUploadInProgress = errors.New("store: uploaded transcript is still being processed")
// Uploaded transcripts are PRIVATE per-user content, stored in their own // Uploaded transcripts are PRIVATE per-user content, stored in their own
// RLS-scoped table (migration 017) — never in the shared `transcripts` (ADR-021). // RLS-scoped table (migration 017) — never in the shared `transcripts` (ADR-021).
// Every method goes through withUser, so row-level security confines it to the // Every method goes through withUser, so row-level security confines it to the
@@ -80,12 +84,14 @@ func (s *Store) SaveProgress(ctx context.Context, userID, id string, steps []dom
WHERE id = $1 AND user_id = $2`, b, at) WHERE id = $1 AND user_id = $2`, b, at)
} }
// Heartbeat stamps the upload as alive now. It does not touch progress_at: alive says // Heartbeat stamps the upload as alive at the given time. It does not touch progress_at:
// the service is still reporting, not that work is moving (ADR-035). // alive says the service is still reporting, not that work is moving (ADR-035). The time
func (s *Store) Heartbeat(ctx context.Context, userID, id string) error { // comes from the application clock, the one progress_at and the page's health check use,
// so a database clock that disagrees cannot make a healthy job look lost (#51).
func (s *Store) Heartbeat(ctx context.Context, userID, id string, at time.Time) error {
return s.updateUpload(ctx, userID, id, "heartbeat", return s.updateUpload(ctx, userID, id, "heartbeat",
`UPDATE uploaded_transcripts SET alive_at = NOW() `UPDATE uploaded_transcripts SET alive_at = $3
WHERE id = $1 AND user_id = $2`) WHERE id = $1 AND user_id = $2`, at)
} }
// FailUpload moves a queued or transcribing upload to failed with a typed reason. A // FailUpload moves a queued or transcribing upload to failed with a typed reason. A
@@ -101,6 +107,33 @@ func (s *Store) FailUpload(ctx context.Context, userID, id string, kind domain.U
string(kind)) string(kind))
} }
// DeleteUpload removes one of the caller's finished (done or failed) uploads. An upload
// still queued or transcribing is refused with ErrUploadInProgress and left untouched; one
// that is not the caller's is ErrUploadNotFound, as for every other write (#37).
func (s *Store) DeleteUpload(ctx context.Context, userID, id string) error {
return s.withUser(ctx, userID, func(tx pgx.Tx) error {
ct, err := tx.Exec(ctx,
`DELETE FROM uploaded_transcripts
WHERE id = $1 AND user_id = $2 AND status IN ('done', 'failed')`, id, userID)
if err != nil {
return fmt.Errorf("store: delete upload: %w", err)
}
if ct.RowsAffected() > 0 {
return nil
}
var exists bool
if err := tx.QueryRow(ctx,
`SELECT EXISTS (SELECT 1 FROM uploaded_transcripts WHERE id = $1 AND user_id = $2)`,
id, userID).Scan(&exists); err != nil {
return fmt.Errorf("store: delete upload: %w", err)
}
if exists {
return ErrUploadInProgress
}
return ErrUploadNotFound
})
}
// updateUpload runs one UPDATE keyed on ($1 = id, $2 = userID) plus extra args, // updateUpload runs one UPDATE keyed on ($1 = id, $2 = userID) plus extra args,
// and maps "no row touched" to ErrUploadNotFound — which is what a write to another // and maps "no row touched" to ErrUploadNotFound — which is what a write to another
// user's upload looks like under RLS (the row is invisible, not forbidden). // user's upload looks like under RLS (the row is invisible, not forbidden).
@@ -0,0 +1,68 @@
package store_test
// Deleting one recording (#37). A finished or failed upload can be removed by its owner;
// one that is queued or transcribing is refused until it ends, because the worker would
// otherwise write its result to a row that is gone (D3 in the sprint plan).
import (
"context"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"git.d-ma.be/mathias/tapir/internal/adapters/store"
"git.d-ma.be/mathias/tapir/internal/domain"
)
func TestDeleteUpload_RemovesAFinishedOrFailedUploadOfTheOwner(t *testing.T) {
for name, finish := range map[string]func(*store.Store, string) error{
"done": func(s *store.Store, id string) error {
return s.CompleteUpload(context.Background(), userA, id, domain.Transcription{Text: "synthetic", Endpoint: "iguana/kb-whisper"})
},
"failed": func(s *store.Store, id string) error {
return s.FailUpload(context.Background(), userA, id, domain.UploadErrTranscriptionFailed)
},
} {
t.Run(name, func(t *testing.T) {
s := newStore(t)
id := startedUpload(t, s)
require.NoError(t, s.StartTranscribing(context.Background(), userA, id))
require.NoError(t, finish(s, id))
require.NoError(t, s.DeleteUpload(context.Background(), userA, id))
_, err := s.GetUpload(context.Background(), userA, id)
assert.ErrorIs(t, err, store.ErrUploadNotFound)
})
}
}
func TestDeleteUpload_RefusesAnUploadThatIsStillQueuedOrTranscribingAndKeepsTheRow(t *testing.T) {
s := newStore(t)
ctx := context.Background()
queued := startedUpload(t, s)
running, err := s.CreateUpload(ctx, userA, "running", false)
require.NoError(t, err)
require.NoError(t, s.StartTranscribing(ctx, userA, running))
for _, id := range []string{queued, running} {
assert.ErrorIs(t, s.DeleteUpload(ctx, userA, id), store.ErrUploadInProgress)
_, err := s.GetUpload(ctx, userA, id)
assert.NoError(t, err, "a refused delete must leave the row")
}
}
func TestDeleteUpload_AnotherUsersUploadIsNotFoundAndSurvives(t *testing.T) {
s := newStore(t)
ctx := context.Background()
id := startedUpload(t, s)
require.NoError(t, s.StartTranscribing(ctx, userA, id))
require.NoError(t, s.FailUpload(ctx, userA, id, domain.UploadErrTranscriptionFailed))
assert.ErrorIs(t, s.DeleteUpload(ctx, userB, id), store.ErrUploadNotFound,
"existence is not disclosed to a non-owner")
_, err := s.GetUpload(ctx, userA, id)
assert.NoError(t, err, "B's delete must not have removed A's row")
}
@@ -68,13 +68,13 @@ func TestUploadProgress_AHeartbeatMovesAliveAtAndNotProgressAt(t *testing.T) {
ctx := context.Background() ctx := context.Background()
at := time.Date(2026, 10, 2, 8, 30, 0, 0, time.UTC) at := time.Date(2026, 10, 2, 8, 30, 0, 0, time.UTC)
require.NoError(t, s.SaveProgress(ctx, userA, id, []domain.ProgressStep{{Key: "audio", Label: "a", State: domain.StepRunning}}, at)) require.NoError(t, s.SaveProgress(ctx, userA, id, []domain.ProgressStep{{Key: "audio", Label: "a", State: domain.StepRunning}}, at))
before := time.Now().Add(-time.Second) beatAt := at.Add(90 * time.Second) // deliberately not the database's NOW()
require.NoError(t, s.Heartbeat(ctx, userA, id)) require.NoError(t, s.Heartbeat(ctx, userA, id, beatAt))
got, err := s.GetUpload(ctx, userA, id) got, err := s.GetUpload(ctx, userA, id)
require.NoError(t, err) require.NoError(t, err)
assert.True(t, got.AliveAt.After(before), "the heartbeat must stamp the row with now") assert.True(t, got.AliveAt.Equal(beatAt), "the heartbeat is stamped with the application's clock, the same one progress_at uses, not the database's NOW() (#51)")
assert.True(t, got.ProgressAt.Equal(at), "alive is not progress: a heartbeat must not move progress_at") assert.True(t, got.ProgressAt.Equal(at), "alive is not progress: a heartbeat must not move progress_at")
} }
@@ -99,7 +99,7 @@ func TestUploadProgress_ByAnotherUserChangesNothing(t *testing.T) {
ctx := context.Background() ctx := context.Background()
errSave := s.SaveProgress(ctx, userB, id, []domain.ProgressStep{{Key: "x", Label: "X", State: domain.StepDone}}, time.Now()) errSave := s.SaveProgress(ctx, userB, id, []domain.ProgressStep{{Key: "x", Label: "X", State: domain.StepDone}}, time.Now())
errBeat := s.Heartbeat(ctx, userB, id) errBeat := s.Heartbeat(ctx, userB, id, time.Now())
assert.ErrorIs(t, errSave, store.ErrUploadNotFound) assert.ErrorIs(t, errSave, store.ErrUploadNotFound)
assert.ErrorIs(t, errBeat, store.ErrUploadNotFound) assert.ErrorIs(t, errBeat, store.ErrUploadNotFound)
+22 -6
View File
@@ -37,12 +37,13 @@ type Reporter interface {
// is for people, so a failed write is logged and never stops the job. // is for people, so a failed write is logged and never stops the job.
type Sink interface { type Sink interface {
SaveProgress(ctx context.Context, steps []domain.ProgressStep, at time.Time) error SaveProgress(ctx context.Context, steps []domain.ProgressStep, at time.Time) error
Beat(ctx context.Context) error Beat(ctx context.Context, at time.Time) error
} }
const ( const (
defaultMinSave = 2 * time.Second defaultMinSave = 2 * time.Second
logAdvanceEach = time.Minute defaultSaveTimeout = 5 * time.Second
logAdvanceEach = time.Minute
) )
// Option tunes a Tracker. // Option tunes a Tracker.
@@ -55,6 +56,14 @@ func WithClock(now func() time.Time) Option { return func(t *Tracker) { t.now =
// Done, Fail and Finish always save. // Done, Fail and Finish always save.
func WithMinSave(d time.Duration) Option { return func(t *Tracker) { t.minSave = d } } func WithMinSave(d time.Duration) Option { return func(t *Tracker) { t.minSave = d } }
// WithSaveTimeout bounds one save or heartbeat write. Progress is cosmetic, so a database
// that stops answering costs a dropped write rather than an unbounded stall. It is still a
// delay: saves run under the tracker's lock, which Begin and Advance (called by the engines)
// need, so each transition can wait up to this long, and a hung database slows a job by
// that much per call, out of the job's own time budget (#51). A dropped final save is not
// retried; the row keeps its last saved step.
func WithSaveTimeout(d time.Duration) Option { return func(t *Tracker) { t.saveTimeout = d } }
// Tracker implements Reporter for one job. // Tracker implements Reporter for one job.
type Tracker struct { type Tracker struct {
ctx context.Context ctx context.Context
@@ -63,6 +72,8 @@ type Tracker struct {
now func() time.Time now func() time.Time
minSave time.Duration minSave time.Duration
saveTimeout time.Duration
mu sync.Mutex mu sync.Mutex
steps []domain.ProgressStep steps []domain.ProgressStep
lastAdvanceSave time.Time lastAdvanceSave time.Time
@@ -72,7 +83,7 @@ type Tracker struct {
// NewTracker plans a job's phases. ctx is used for the saves, so it should outlive the // NewTracker plans a job's phases. ctx is used for the saves, so it should outlive the
// job's own timeout: a job that timed out still gets its failure recorded. // job's own timeout: a job that timed out still gets its failure recorded.
func NewTracker(ctx context.Context, sink Sink, log *slog.Logger, plan []Planned, opts ...Option) *Tracker { func NewTracker(ctx context.Context, sink Sink, log *slog.Logger, plan []Planned, opts ...Option) *Tracker {
t := &Tracker{ctx: ctx, sink: sink, log: log, now: time.Now, minSave: defaultMinSave} t := &Tracker{ctx: ctx, sink: sink, log: log, now: time.Now, minSave: defaultMinSave, saveTimeout: defaultSaveTimeout}
for _, o := range opts { for _, o := range opts {
o(t) o(t)
} }
@@ -212,7 +223,10 @@ func (t *Tracker) Heartbeat(ctx context.Context, beatEvery, stillEvery time.Dura
case <-ctx.Done(): case <-ctx.Done():
return return
case <-beat.C: case <-beat.C:
if err := t.sink.Beat(ctx); err != nil && ctx.Err() == nil { beatCtx, cancel := context.WithTimeout(ctx, t.saveTimeout)
err := t.sink.Beat(beatCtx, t.now())
cancel()
if err != nil && ctx.Err() == nil {
t.log.Warn("progress: heartbeat not saved", "err", err) t.log.Warn("progress: heartbeat not saved", "err", err)
} }
case <-still.C: case <-still.C:
@@ -254,7 +268,9 @@ func (t *Tracker) find(key string) *domain.ProgressStep {
// dropped: progress is cosmetic and must never fail the job it describes. // dropped: progress is cosmetic and must never fail the job it describes.
func (t *Tracker) save() { func (t *Tracker) save() {
snapshot := append([]domain.ProgressStep(nil), t.steps...) snapshot := append([]domain.ProgressStep(nil), t.steps...)
if err := t.sink.SaveProgress(t.ctx, snapshot, t.now()); err != nil && t.ctx.Err() == nil { ctx, cancel := context.WithTimeout(t.ctx, t.saveTimeout)
defer cancel()
if err := t.sink.SaveProgress(ctx, snapshot, t.now()); err != nil && t.ctx.Err() == nil {
t.log.Warn("progress: not saved", "err", err) t.log.Warn("progress: not saved", "err", err)
} }
} }
+154
View File
@@ -0,0 +1,154 @@
package progress_test
// Observes what a database that stops answering does to the job it reports on (#51):
// progress is cosmetic, so a save that never returns must not stall the engine calls that
// report through the tracker, nor the heartbeat that says the process is alive.
import (
"bytes"
"context"
"log/slog"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/stretchr/testify/require"
"git.d-ma.be/mathias/tapir/internal/domain"
"git.d-ma.be/mathias/tapir/internal/progress"
)
// stallSink never answers until its context ends, like a database connection that hung.
type stallSink struct{ beats atomic.Int32 }
func (s *stallSink) SaveProgress(ctx context.Context, _ []domain.ProgressStep, _ time.Time) error {
<-ctx.Done()
return ctx.Err()
}
func (s *stallSink) Beat(ctx context.Context, _ time.Time) error {
s.beats.Add(1)
<-ctx.Done()
return ctx.Err()
}
type lockedBuf struct {
mu sync.Mutex
b bytes.Buffer
}
func (l *lockedBuf) Write(p []byte) (int, error) {
l.mu.Lock()
defer l.mu.Unlock()
return l.b.Write(p)
}
func (l *lockedBuf) String() string {
l.mu.Lock()
defer l.mu.Unlock()
return l.b.String()
}
func stallTracker(sink progress.Sink, buf *lockedBuf) *progress.Tracker {
log := slog.New(slog.NewTextHandler(buf, nil))
return progress.NewTracker(context.Background(), sink, log,
[]progress.Planned{{Key: "a", Label: "A"}}, progress.WithSaveTimeout(50*time.Millisecond))
}
func TestTracker_AHungSaveDoesNotBlockTheReporterPastItsTimeout(t *testing.T) {
buf := &lockedBuf{}
tr := stallTracker(&stallSink{}, buf)
done := make(chan struct{})
go func() { tr.Begin("a", 0); tr.Advance("a", 1); tr.Done("a"); close(done) }()
select {
case <-done:
case <-time.After(3 * time.Second):
t.Fatal("the reporter was blocked by a save that never returned")
}
require.Contains(t, buf.String(), "progress: not saved", "a dropped save is logged")
}
func TestTracker_AHungBeatDoesNotStopTheHeartbeat(t *testing.T) {
sink := &stallSink{}
tr := stallTracker(sink, &lockedBuf{})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go tr.Heartbeat(ctx, 5*time.Millisecond, time.Hour)
require.Eventually(t, func() bool { return sink.beats.Load() >= 3 }, 3*time.Second, 5*time.Millisecond,
"each beat is abandoned at its timeout and the next one is tried")
}
func TestTracker_TheStillRunningLineIsNotHeldUpBehindAHungSave(t *testing.T) {
buf := &lockedBuf{}
tr := stallTracker(&stallSink{}, buf)
go tr.Begin("a", 0) // holds the tracker's lock inside a save that hangs until its timeout
time.Sleep(10 * time.Millisecond)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go tr.Heartbeat(ctx, time.Hour, 5*time.Millisecond)
require.Eventually(t, func() bool { return strings.Contains(buf.String(), "still running") }, 3*time.Second, 5*time.Millisecond)
}
// deadlineSink records how long each write was given, and answers at once.
type deadlineSink struct {
mu sync.Mutex
saves []time.Duration
beats []time.Duration
noDLOK bool
}
func (d *deadlineSink) record(ctx context.Context, into *[]time.Duration) {
d.mu.Lock()
defer d.mu.Unlock()
dl, ok := ctx.Deadline()
if !ok {
d.noDLOK = true
return
}
*into = append(*into, time.Until(dl))
}
func (d *deadlineSink) SaveProgress(ctx context.Context, _ []domain.ProgressStep, _ time.Time) error {
d.record(ctx, &d.saves)
return nil
}
func (d *deadlineSink) Beat(ctx context.Context, _ time.Time) error {
d.record(ctx, &d.beats)
return nil
}
// The production value is the one the worker gets: it builds a Tracker without options.
// A zero or tiny default would drop every write, a huge one would bound nothing.
func TestTracker_TheDefaultSaveAndBeatDeadlineIsAFewSecondsAway(t *testing.T) {
sink := &deadlineSink{}
tr := progress.NewTracker(context.Background(), sink, slog.New(slog.NewTextHandler(&lockedBuf{}, nil)),
[]progress.Planned{{Key: "a", Label: "A"}})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go tr.Heartbeat(ctx, 5*time.Millisecond, time.Hour)
tr.Begin("a", 0)
require.Eventually(t, func() bool {
sink.mu.Lock()
defer sink.mu.Unlock()
return len(sink.beats) > 0
}, 3*time.Second, 5*time.Millisecond)
sink.mu.Lock()
defer sink.mu.Unlock()
require.False(t, sink.noDLOK, "a write was given no deadline at all")
require.NotEmpty(t, sink.saves)
for _, d := range append(append([]time.Duration{}, sink.saves...), sink.beats...) {
require.Greater(t, d, time.Second, "the deadline is already (nearly) spent when the write starts")
require.LessOrEqual(t, d, 5*time.Second, "the default bound is 5 s")
}
}
+26 -7
View File
@@ -17,11 +17,12 @@ import (
// fakeSink records every saved snapshot and every heartbeat. // fakeSink records every saved snapshot and every heartbeat.
type fakeSink struct { type fakeSink struct {
mu sync.Mutex mu sync.Mutex
saves [][]domain.ProgressStep saves [][]domain.ProgressStep
at []time.Time at []time.Time
beats int beats int
err error beatAt []time.Time
err error
} }
func (f *fakeSink) SaveProgress(_ context.Context, steps []domain.ProgressStep, at time.Time) error { func (f *fakeSink) SaveProgress(_ context.Context, steps []domain.ProgressStep, at time.Time) error {
@@ -32,10 +33,11 @@ func (f *fakeSink) SaveProgress(_ context.Context, steps []domain.ProgressStep,
return f.err return f.err
} }
func (f *fakeSink) Beat(context.Context) error { func (f *fakeSink) Beat(_ context.Context, at time.Time) error {
f.mu.Lock() f.mu.Lock()
defer f.mu.Unlock() defer f.mu.Unlock()
f.beats++ f.beats++
f.beatAt = append(f.beatAt, at)
return f.err return f.err
} }
@@ -297,7 +299,7 @@ func (r *retainingSink) SaveProgress(_ context.Context, steps []domain.ProgressS
r.got = append(r.got, steps) r.got = append(r.got, steps)
return nil return nil
} }
func (r *retainingSink) Beat(context.Context) error { return nil } func (r *retainingSink) Beat(context.Context, time.Time) error { return nil }
func TestTracker_ASnapshotHandedToTheSinkIsNotChangedByLaterTransitions(t *testing.T) { func TestTracker_ASnapshotHandedToTheSinkIsNotChangedByLaterTransitions(t *testing.T) {
sink := &retainingSink{} sink := &retainingSink{}
@@ -400,3 +402,20 @@ func TestTracker_AnnotateAttachesANoteToAStepAndSavesItAtOnce(t *testing.T) {
require.Equal(t, "20 min of the 79 the file claimed", sink.last(t)[0].Note) require.Equal(t, "20 min of the 79 the file claimed", sink.last(t)[0].Note)
require.NotPanics(t, func() { tr.Annotate("no-such-step", "x") }) require.NotPanics(t, func() { tr.Annotate("no-such-step", "x") })
} }
func TestTracker_BeatIsStampedWithTheTrackersClock(t *testing.T) {
// alive_at used the database's NOW() while the page compared it with the application
// clock; a skewed database clock showed a healthy job as lost (#51).
tr, sink, clk, _ := newTracker(t)
clk.tick(42 * time.Second)
want := clk.now()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go tr.Heartbeat(ctx, 5*time.Millisecond, time.Hour)
require.Eventually(t, func() bool { return sink.beatCount() >= 1 }, 2*time.Second, 5*time.Millisecond)
sink.mu.Lock()
defer sink.mu.Unlock()
require.True(t, sink.beatAt[0].Equal(want), "beat stamped %v, the tracker's clock says %v", sink.beatAt[0], want)
}
+48
View File
@@ -0,0 +1,48 @@
package uploads
// Observes what jobSink hands the store: the heartbeat's timestamp is the one the tracker
// stamped (its application clock), not a second reading taken here (#51).
import (
"context"
"testing"
"time"
"git.d-ma.be/mathias/tapir/internal/domain"
)
type atStore struct{ beatAt, saveAt time.Time }
func (*atStore) StartTranscribing(context.Context, string, string) error { return nil }
func (*atStore) CompleteUpload(context.Context, string, string, domain.Transcription) error {
return nil
}
func (*atStore) FailUpload(context.Context, string, string, domain.UploadErrorKind) error { return nil }
func (s *atStore) SaveProgress(_ context.Context, _, _ string, _ []domain.ProgressStep, at time.Time) error {
s.saveAt = at
return nil
}
func (s *atStore) Heartbeat(_ context.Context, _, _ string, at time.Time) error {
s.beatAt = at
return nil
}
func TestJobSink_ForwardsTheTrackersTimestampToTheStoreUnchanged(t *testing.T) {
st := &atStore{}
sink := jobSink{st: st, userID: "u", id: "i"}
stamp := time.Date(2020, 1, 2, 3, 4, 5, 0, time.UTC)
if err := sink.Beat(context.Background(), stamp); err != nil {
t.Fatal(err)
}
if err := sink.SaveProgress(context.Background(), nil, stamp); err != nil {
t.Fatal(err)
}
if !st.beatAt.Equal(stamp) {
t.Errorf("heartbeat reached the store stamped %v, want the tracker's %v", st.beatAt, stamp)
}
if !st.saveAt.Equal(stamp) {
t.Errorf("progress save reached the store stamped %v, want the tracker's %v", st.saveAt, stamp)
}
}
+4 -2
View File
@@ -47,7 +47,7 @@ type WorkerStore interface {
CompleteUpload(ctx context.Context, userID, id string, t domain.Transcription) error CompleteUpload(ctx context.Context, userID, id string, t domain.Transcription) error
FailUpload(ctx context.Context, userID, id string, kind domain.UploadErrorKind) error FailUpload(ctx context.Context, userID, id string, kind domain.UploadErrorKind) error
SaveProgress(ctx context.Context, userID, id string, steps []domain.ProgressStep, at time.Time) error SaveProgress(ctx context.Context, userID, id string, steps []domain.ProgressStep, at time.Time) error
Heartbeat(ctx context.Context, userID, id string) error Heartbeat(ctx context.Context, userID, id string, at time.Time) error
} }
// WorkerOption tunes a Worker. // WorkerOption tunes a Worker.
@@ -91,7 +91,9 @@ func (s jobSink) SaveProgress(ctx context.Context, steps []domain.ProgressStep,
return s.st.SaveProgress(ctx, s.userID, s.id, steps, at) return s.st.SaveProgress(ctx, s.userID, s.id, steps, at)
} }
func (s jobSink) Beat(ctx context.Context) error { return s.st.Heartbeat(ctx, s.userID, s.id) } func (s jobSink) Beat(ctx context.Context, at time.Time) error {
return s.st.Heartbeat(ctx, s.userID, s.id, at)
}
// Worker transcribes accepted uploads in-process, ONE AT A TIME (ADR-033). It is // Worker transcribes accepted uploads in-process, ONE AT A TIME (ADR-033). It is
// not a second deployable: it shares the web process's lifetime, which is why it // not a second deployable: it shares the web process's lifetime, which is why it
+1 -1
View File
@@ -70,7 +70,7 @@ func (s *wStore) SaveProgress(_ context.Context, _, _ string, steps []domain.Pro
return nil return nil
} }
func (s *wStore) Heartbeat(context.Context, string, string) error { func (s *wStore) Heartbeat(context.Context, string, string, time.Time) error {
time.Sleep(s.beatDelay) time.Sleep(s.beatDelay)
s.mu.Lock() s.mu.Lock()
defer s.mu.Unlock() defer s.mu.Unlock()
+14
View File
@@ -125,6 +125,20 @@ func TestCrossOrigin_AForeignPostIsRefusedOnEachListedStateChangingPath(t *testi
} }
} }
func TestCrossOrigin_ARefusedRecordingDeletePostLeavesTheRecordingInPlace(t *testing.T) {
for name, h := range foreign {
t.Run(name, func(t *testing.T) {
app, _ := newUploadApp(t, audioOK, 200<<20)
id := seedUpload(t, userID, "m", "done", "Synthetic transcript text.")
rec := do(t, app, withHeaders(httptest.NewRequest(http.MethodPost, "/uploads/"+id+"/delete", nil), h))
assert.Equal(t, http.StatusForbidden, rec.Code)
assert.True(t, uploadExists(t, id), "a refused post must not delete the recording")
})
}
}
// A login redirect comes back to /auth/callback as a cross-site GET navigation, and // A login redirect comes back to /auth/callback as a cross-site GET navigation, and
// health probes carry no headers: safe methods must never be refused on origin grounds. // health probes carry no headers: safe methods must never be refused on origin grounds.
func TestCrossOrigin_ACrossSiteGETIsNotRefused(t *testing.T) { func TestCrossOrigin_ACrossSiteGETIsNotRefused(t *testing.T) {
+5 -1
View File
@@ -59,6 +59,9 @@ type Store interface {
// non-owner's id is ErrUploadNotFound. // non-owner's id is ErrUploadNotFound.
GetUpload(ctx context.Context, userID, id string) (*domain.UploadedTranscript, error) GetUpload(ctx context.Context, userID, id string) (*domain.UploadedTranscript, error)
ListUploads(ctx context.Context, userID string) ([]domain.UploadedTranscript, error) ListUploads(ctx context.Context, userID string) ([]domain.UploadedTranscript, error)
// DeleteUpload removes a finished upload; ErrUploadInProgress while it is still being
// processed, ErrUploadNotFound for a non-owner.
DeleteUpload(ctx context.Context, userID, id string) error
// SetTranscriptStatus clears or updates a video's transcript backoff state. // SetTranscriptStatus clears or updates a video's transcript backoff state.
// Used by handleRetryNow to clear rate_limited_at before immediate processing. // Used by handleRetryNow to clear rate_limited_at before immediate processing.
@@ -200,6 +203,7 @@ func (a *App) Router() http.Handler {
app.HandleFunc("POST /uploads", a.handleUploadPost) app.HandleFunc("POST /uploads", a.handleUploadPost)
app.HandleFunc("GET /uploads/{id}", a.handleUploadDetail) app.HandleFunc("GET /uploads/{id}", a.handleUploadDetail)
app.HandleFunc("GET /uploads/{id}/status", a.handleUploadStatus) app.HandleFunc("GET /uploads/{id}/status", a.handleUploadStatus)
app.HandleFunc("POST /uploads/{id}/delete", a.handleUploadDelete)
// Chat over the uploaded transcript reuses the ADR-027 Chatter; mounted only // Chat over the uploaded transcript reuses the ADR-027 Chatter; mounted only
// when one is wired. // when one is wired.
if a.Chat != nil { if a.Chat != nil {
@@ -309,7 +313,7 @@ func (a *App) handleList(w http.ResponseWriter, r *http.Request) {
a.render(w, r, summaryList(buckets, hasConnected, autoSummarize)) a.render(w, r, summaryList(buckets, hasConnected, autoSummarize))
return return
} }
a.render(w, r, ListPage(buckets, f, stats, takeFlash(w, r), hasConnected, channels, autoSummarize)) a.render(w, r, ListPage(buckets, f, stats, takeFlash(w, r), hasConnected, channels, autoSummarize, a.Uploads != nil))
} }
// handleDetail renders one summary in full (highlights, takeaways, action group). // handleDetail renders one summary in full (highlights, takeaways, action group).
+1 -1
View File
@@ -66,7 +66,7 @@ func TestParseYouTubeVideoID(t *testing.T) {
func TestListPageShowsPasteFormOnlyWhenConnected(t *testing.T) { func TestListPageShowsPasteFormOnlyWhenConnected(t *testing.T) {
render := func(connected bool) string { render := func(connected bool) string {
var buf bytes.Buffer var buf bytes.Buffer
if err := ListPage(listBuckets{}, Filter{}, PipelineStats{}, "", connected, nil, true).Render(context.Background(), &buf); err != nil { if err := ListPage(listBuckets{}, Filter{}, PipelineStats{}, "", connected, nil, true, false).Render(context.Background(), &buf); err != nil {
t.Fatalf("render: %v", err) t.Fatalf("render: %v", err)
} }
return buf.String() return buf.String()
+57
View File
@@ -18,6 +18,7 @@ import (
"git.d-ma.be/mathias/tapir/internal/adapters/chat" "git.d-ma.be/mathias/tapir/internal/adapters/chat"
"git.d-ma.be/mathias/tapir/internal/adapters/store" "git.d-ma.be/mathias/tapir/internal/adapters/store"
"git.d-ma.be/mathias/tapir/internal/config"
"git.d-ma.be/mathias/tapir/internal/domain" "git.d-ma.be/mathias/tapir/internal/domain"
"git.d-ma.be/mathias/tapir/internal/errkind" "git.d-ma.be/mathias/tapir/internal/errkind"
"git.d-ma.be/mathias/tapir/internal/uploads" "git.d-ma.be/mathias/tapir/internal/uploads"
@@ -209,6 +210,32 @@ func (a *App) handleUploadStatus(w http.ResponseWriter, r *http.Request) {
a.render(w, r, uploadStatus(*u, a.Chat != nil, a.now())) a.render(w, r, uploadStatus(*u, a.Chat != nil, a.now()))
} }
// handleUploadDelete removes one of the caller's finished recordings and returns to the
// list. Another user's id is a 404 like every other upload route; a recording still being
// transcribed is refused (409) rather than deleted under the worker (#37).
func (a *App) handleUploadDelete(w http.ResponseWriter, r *http.Request) {
userID, ok := a.currentUserID(w, r)
if !ok {
return
}
id := r.PathValue("id")
if !uuidShape.MatchString(id) {
http.NotFound(w, r)
return
}
switch err := a.Store.DeleteUpload(r.Context(), userID, id); {
case err == nil:
http.Redirect(w, r, "/uploads", http.StatusSeeOther)
case errors.Is(err, store.ErrUploadNotFound):
http.NotFound(w, r)
case errors.Is(err, store.ErrUploadInProgress):
a.renderUploadsPage(w, r, userID, http.StatusConflict,
"This recording is still being transcribed. You can delete it once it has finished.")
default:
a.serverError(w, r, "delete upload", err)
}
}
func (a *App) loadOwnedUpload(w http.ResponseWriter, r *http.Request) (*domain.UploadedTranscript, bool) { func (a *App) loadOwnedUpload(w http.ResponseWriter, r *http.Request) (*domain.UploadedTranscript, bool) {
userID, ok := a.currentUserID(w, r) userID, ok := a.currentUserID(w, r)
if !ok { if !ok {
@@ -243,6 +270,36 @@ func titleFromFilename(name string) string {
return t return t
} }
// cloudProvider names the cloud service that produced a transcript, or false when it was
// served locally (or by an unknown endpoint). "Local" is config.IsLocalModel: the same
// definition the start-up validation uses, not a second string match here (#52).
//
// The endpoint is one alias, except for the two-engine path (ADR-034), which writes
// "<base> + <alt>, reconciled by <model>" (ensemble.go). That shape is taken apart exactly
// and each alias is judged whole, so a composite of local models is local, and anything
// else (a cloud alias, one with no provider prefix, one with a space or a connector word in
// it) is not. A string that is not that shape is one alias.
func cloudProvider(endpoint string) (string, bool) {
endpoint = strings.TrimSpace(endpoint)
if endpoint == "" {
return "", false
}
aliases := []string{endpoint}
if engines, reconciler, ok := strings.Cut(endpoint, ", reconciled by "); ok {
aliases = append(strings.Split(engines, " + "), reconciler)
}
for _, a := range aliases {
if a = strings.TrimSpace(a); !config.IsLocalModel(a) {
provider, _, _ := strings.Cut(a, "/")
if provider = strings.TrimSpace(provider); provider == "" {
provider = a
}
return provider, true
}
}
return "", false
}
// hasNote reports whether any phase carries a note. A note is the only warning for a // hasNote reports whether any phase carries a note. A note is the only warning for a
// recording that is shorter than its file claimed, so a record that has one is shown open. // recording that is shorter than its file claimed, so a record that has one is shown open.
func hasNote(steps []domain.ProgressStep) bool { func hasNote(steps []domain.ProgressStep) bool {
+22
View File
@@ -66,6 +66,12 @@ templ uploadStatus(u domain.UploadedTranscript, chatEnabled bool, now time.Time)
Transcribed { uploads.FormatDuration(time.Duration(u.AudioSeconds * float64(time.Second))) } Transcribed { uploads.FormatDuration(time.Duration(u.AudioSeconds * float64(time.Second))) }
of audio by <span class="chip">{ u.Endpoint }</span>. of audio by <span class="chip">{ u.Endpoint }</span>.
</p> </p>
if provider, cloud := cloudProvider(u.Endpoint); cloud {
<p class="cloud-note">
This recording's audio left the homelab: it was transcribed by a cloud service ({ provider })
because the local engine could not.
</p>
}
@reconciliationNote(u.Reconciliation) @reconciliationNote(u.Reconciliation)
if len(u.Progress) > 0 { if len(u.Progress) > 0 {
<details class="what-was-done" if hasNote(u.Progress) { open }> <details class="what-was-done" if hasNote(u.Progress) { open }>
@@ -80,6 +86,7 @@ templ uploadStatus(u domain.UploadedTranscript, chatEnabled bool, now time.Time)
<summary>Transcript</summary> <summary>Transcript</summary>
<pre class="transcript">{ u.Content }</pre> <pre class="transcript">{ u.Content }</pre>
</details> </details>
@deleteRecording(u.ID)
</div> </div>
case domain.UploadFailed: case domain.UploadFailed:
<div class="upload-status upload-failed flash flash-error" role="alert"> <div class="upload-status upload-failed flash flash-error" role="alert">
@@ -91,6 +98,7 @@ templ uploadStatus(u domain.UploadedTranscript, chatEnabled bool, now time.Time)
@progressSteps(u.Progress, now, false) @progressSteps(u.Progress, now, false)
</details> </details>
} }
@deleteRecording(u.ID)
</div> </div>
default: default:
<div <div
@@ -116,6 +124,20 @@ templ uploadStatus(u domain.UploadedTranscript, chatEnabled bool, now time.Time)
} }
} }
// deleteRecording is the confirm-then-delete control, shown only once a recording has
// finished (done or failed): deleting one that is still being processed is refused.
templ deleteRecording(id string) {
<details class="confirm-delete">
<summary class="btn-danger">Delete this recording…</summary>
<div class="confirm-body">
<p>This permanently deletes the recording and its transcript. Are you sure?</p>
<form method="post" action={ templ.SafeURL("/uploads/" + id + "/delete") }>
<button type="submit" class="btn-danger">Yes, permanently delete it</button>
</form>
</div>
</details>
}
// UploadChatPage is the no-JS full page for chatting with an uploaded transcript: // UploadChatPage is the no-JS full page for chatting with an uploaded transcript:
// the same chat panel as a video's (ADR-027), scoped to this recording. // the same chat panel as a video's (ADR-027), scoped to this recording.
templ UploadChatPage(u domain.UploadedTranscript, v chatView) { templ UploadChatPage(u domain.UploadedTranscript, v chatView) {
+140
View File
@@ -0,0 +1,140 @@
package web_test
// A transcript produced by the cloud fallback means this recording's audio left the
// homelab, which the reader is told in plain words (#52). "Local" is config.IsLocalModel:
// one definition, not a second string match in the view.
import (
"context"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func setEndpoint(t *testing.T, id, endpoint string) {
t.Helper()
_, err := rawPool(t).Exec(context.Background(), `UPDATE uploaded_transcripts SET endpoint = $2 WHERE id = $1`, id, endpoint)
require.NoError(t, err)
}
func TestUploadStatus_ACloudServedTranscriptSaysTheAudioLeftTheHomelab(t *testing.T) {
app, _ := newUploadApp(t, audioOK, 200<<20)
id := seedUpload(t, userID, "m", "done", "Synthetic transcript text.")
setEndpoint(t, id, "berget/kb-whisper")
b := body(t, do(t, app, get("/uploads/"+id+"/status")))
assert.Contains(t, b, "left the homelab")
assert.Contains(t, b, "cloud service (berget)")
}
func TestUploadStatus_ALocallyServedTranscriptCarriesNoCloudNote(t *testing.T) {
for _, endpoint := range []string{"iguana/kb-whisper", "koala/whisper", ""} {
t.Run(endpoint, func(t *testing.T) {
app, _ := newUploadApp(t, audioOK, 200<<20)
id := seedUpload(t, userID, "m", "done", "Synthetic transcript text.")
setEndpoint(t, id, endpoint)
b := body(t, do(t, app, get("/uploads/"+id+"/status")))
assert.NotContains(t, b, "left the homelab")
})
}
}
func TestUploadStatus_AnEndpointThatOnlyLooksLocalIsStillNamedAsCloud(t *testing.T) {
// The prefix alone must not decide it: IsLocalModel refuses path tricks.
app, _ := newUploadApp(t, audioOK, 200<<20)
id := seedUpload(t, userID, "m", "done", "Synthetic transcript text.")
setEndpoint(t, id, "iguana/../berget/kb-whisper")
b := body(t, do(t, app, get("/uploads/"+id+"/status")))
assert.Contains(t, b, "left the homelab")
}
// The two-engine path (ADR-034) writes one composite endpoint string. Every name in it is
// a local model, so the audio never left the homelab and the note must not say it did.
func TestUploadStatus_AReconciledAllLocalEndpointCarriesNoCloudNote(t *testing.T) {
for _, endpoint := range []string{
"iguana/kb-whisper + koala/whisper, reconciled by iguana/gemma4-26b",
"iguana/kb-whisper + koala/whisper, reconciled by iguana/gemma4-26b\n",
} {
t.Run(endpoint, func(t *testing.T) {
app, _ := newUploadApp(t, audioOK, 200<<20)
id := seedUpload(t, userID, "m", "done", "Synthetic transcript text.")
setEndpoint(t, id, endpoint)
b := body(t, do(t, app, get("/uploads/"+id+"/status")))
assert.NotContains(t, b, "left the homelab")
})
}
}
func TestUploadStatus_AReconciledEndpointNamingACloudEngineStillNamesTheCloudProvider(t *testing.T) {
app, _ := newUploadApp(t, audioOK, 200<<20)
id := seedUpload(t, userID, "m", "done", "Synthetic transcript text.")
setEndpoint(t, id, "iguana/kb-whisper + berget/kb-whisper, reconciled by iguana/gemma4-26b")
b := body(t, do(t, app, get("/uploads/"+id+"/status")))
assert.Contains(t, b, "cloud service (berget)")
}
// An alias without a provider prefix is not a local model, wherever it sits in the
// composite: it must never be dropped from the judgement (PR #58 re-review).
func TestUploadStatus_AReconciledEndpointWithAnUnprefixedAliasIsNotCalledLocal(t *testing.T) {
for _, endpoint := range []string{
"iguana/kb-whisper + whisper-1, reconciled by iguana/gemma4-26b",
"iguana/kb-whisper + whisper 1, reconciled by iguana/gemma4-26b",
"iguana/kb-whisper + berget / x, reconciled by iguana/gemma4-26b",
} {
t.Run(endpoint, func(t *testing.T) {
app, _ := newUploadApp(t, audioOK, 200<<20)
id := seedUpload(t, userID, "m", "done", "Synthetic transcript text.")
setEndpoint(t, id, endpoint)
b := body(t, do(t, app, get("/uploads/"+id+"/status")))
assert.Contains(t, b, "left the homelab")
assert.NotContains(t, b, "cloud service ()", "the note names a provider")
})
}
}
// "by", "reconciled" and "+" are connectors only inside the composite format. As a whole
// endpoint, or as an alias within the composite, they are model names and are judged.
func TestUploadStatus_AConnectorWordUsedAsAModelNameIsStillJudged(t *testing.T) {
for _, endpoint := range []string{
"by",
"reconciled",
"+",
"iguana/kb-whisper + by, reconciled by iguana/gemma4-26b",
"iguana/kb-whisper + reconciled, reconciled by iguana/gemma4-26b",
"iguana/kb-whisper + +, reconciled by iguana/gemma4-26b",
"/x", // no provider at all: the note must still name something
} {
t.Run(endpoint, func(t *testing.T) {
app, _ := newUploadApp(t, audioOK, 200<<20)
id := seedUpload(t, userID, "m", "done", "Synthetic transcript text.")
setEndpoint(t, id, endpoint)
b := body(t, do(t, app, get("/uploads/"+id+"/status")))
assert.Contains(t, b, "left the homelab")
assert.NotContains(t, b, "cloud service ()")
})
}
}
func TestUploadStatus_AReconcilerThatIsNotLocalNamesTheCloudProvider(t *testing.T) {
app, _ := newUploadApp(t, audioOK, 200<<20)
id := seedUpload(t, userID, "m", "done", "Synthetic transcript text.")
setEndpoint(t, id, "iguana/kb-whisper + koala/whisper, reconciled by berget/mistral-large")
b := body(t, do(t, app, get("/uploads/"+id+"/status")))
assert.Contains(t, b, "cloud service (berget)")
}
@@ -0,0 +1,85 @@
package web_test
// Deleting one recording (#37). Owner-scoped like every other upload route: another user's
// id is a 404 and the row survives; an upload still being processed is refused.
import (
"context"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func deleteReq(id string) *http.Request {
return httptest.NewRequest(http.MethodPost, "/uploads/"+id+"/delete", nil)
}
func uploadExists(t *testing.T, id string) bool {
t.Helper()
var n int
require.NoError(t, rawPool(t).QueryRow(context.Background(), `SELECT count(*) FROM uploaded_transcripts WHERE id = $1`, id).Scan(&n))
return n == 1
}
func TestUploadDelete_AnOwnerRemovesAFinishedOrFailedRecordingAndLandsOnTheList(t *testing.T) {
for _, status := range []string{"done", "failed"} {
t.Run(status, func(t *testing.T) {
app, _ := newUploadApp(t, audioOK, 200<<20)
id := seedUpload(t, userID, "Roadmap sync", status, "transcription_failed")
rec := do(t, app, deleteReq(id))
assert.Equal(t, http.StatusSeeOther, rec.Code)
assert.Equal(t, "/uploads", rec.Header().Get("Location"))
assert.False(t, uploadExists(t, id))
})
}
}
func TestUploadDelete_AnotherUsersRecordingIs404AndSurvives(t *testing.T) {
app, _ := newUploadApp(t, audioOK, 200<<20)
theirs := seedUpload(t, otherUserID, "Their private meeting", "done", meetingText)
rec := do(t, app, deleteReq(theirs))
assert.Equal(t, http.StatusNotFound, rec.Code)
assert.True(t, uploadExists(t, theirs), "a non-owner's delete must not remove the row")
}
func TestUploadDelete_ARecordingStillBeingTranscribedIsRefusedWith409AndKept(t *testing.T) {
for _, status := range []string{"queued", "transcribing"} {
t.Run(status, func(t *testing.T) {
app, _ := newUploadApp(t, audioOK, 200<<20)
id := seedUpload(t, userID, "m", status, "")
rec := do(t, app, deleteReq(id))
assert.Equal(t, http.StatusConflict, rec.Code)
assert.Contains(t, body(t, rec), "still being transcribed")
assert.True(t, uploadExists(t, id))
})
}
}
func TestUploadDelete_ANonIDIsA404NotAServerError(t *testing.T) {
app, _ := newUploadApp(t, audioOK, 200<<20)
assert.Equal(t, http.StatusNotFound, do(t, app, deleteReq("not-a-uuid")).Code)
}
func TestUploadPage_OffersDeleteOnlyForAFinishedOrFailedRecording(t *testing.T) {
for status, want := range map[string]bool{"done": true, "failed": true, "queued": false, "transcribing": false} {
t.Run(status, func(t *testing.T) {
app, _ := newUploadApp(t, audioOK, 200<<20)
id := seedUpload(t, userID, "m", status, "transcription_failed")
b := body(t, do(t, app, get("/uploads/"+id)))
assert.Equal(t, want, strings.Contains(b, `action="/uploads/`+id+`/delete"`))
})
}
}
+19
View File
@@ -0,0 +1,19 @@
package web_test
// /uploads was mounted but nothing linked to it (#37). The shared Layout has no access to
// the App, so the entry point is the home list, shown only when uploads are mounted.
import (
"testing"
"github.com/stretchr/testify/assert"
)
func TestHomeList_LinksToRecordingsOnlyWhenUploadsAreMounted(t *testing.T) {
with, _ := newUploadApp(t, audioOK, 200<<20)
without := newApp(t)
assert.Contains(t, body(t, do(t, with, get("/"))), `href="/uploads"`)
assert.NotContains(t, body(t, do(t, without, get("/"))), `href="/uploads"`,
"a link to a route that is not mounted would be a dead end")
}
File diff suppressed because it is too large Load Diff
+4 -1
View File
@@ -107,9 +107,12 @@ templ flashBanner(code string) {
// #summary-list region; a non-HTMX request renders the whole page. flash carries // #summary-list region; a non-HTMX request renders the whole page. flash carries
// a one-shot notification (e.g. "connected", "registered") surfaced on arrival // a one-shot notification (e.g. "connected", "registered") surfaced on arrival
// after a POST→redirect. // after a POST→redirect.
templ ListPage(b listBuckets, f Filter, stats PipelineStats, flash string, hasConnected bool, channels []string, autoSummarize bool) { templ ListPage(b listBuckets, f Filter, stats PipelineStats, flash string, hasConnected bool, channels []string, autoSummarize bool, uploadsEnabled bool) {
@Layout("Tapir — Summaries") { @Layout("Tapir — Summaries") {
@flashBanner(flash) @flashBanner(flash)
if uploadsEnabled {
<p class="muted"><a href="/uploads">Recordings</a>: upload audio and get a private transcript.</p>
}
if hasConnected { if hasConnected {
@pasteForm() @pasteForm()
} }
File diff suppressed because it is too large Load Diff
@@ -122,6 +122,12 @@ var scenarioCoverage = map[string]string{
"I can ask about my recording and get a grounded answer": "TestUploadChatPost_SendsTheOwnersTranscriptWithTheUploadBudget", "I can ask about my recording and get a grounded answer": "TestUploadChatPost_SendsTheOwnersTranscriptWithTheUploadBudget",
"A long recording is not cut to its first third": "TestAnswerMaxCharsOverridesTheServiceCap", "A long recording is not cut to its first third": "TestAnswerMaxCharsOverridesTheServiceCap",
"A transcript made by the cloud fallback says the audio left the homelab": "TestUploadStatus_ACloudServedTranscriptSaysTheAudioLeftTheHomelab",
"I can delete a finished recording": "TestUploadDelete_AnOwnerRemovesAFinishedOrFailedRecordingAndLandsOnTheList",
"A recording still being transcribed cannot be deleted": "TestUploadDelete_ARecordingStillBeingTranscribedIsRefusedWith409AndKept",
"The home list links to my recordings when uploads are on": "TestHomeList_LinksToRecordingsOnlyWhenUploadsAreMounted",
// audio_upload.feature, ADR-034 // audio_upload.feature, ADR-034
"Two engines transcribe the audio and their disagreements are reconciled": "TestEnsemble_BothEnginesRunInOrderAndTheirDisagreementIsReconciled", "Two engines transcribe the audio and their disagreements are reconciled": "TestEnsemble_BothEnginesRunInOrderAndTheirDisagreementIsReconciled",
"Trivial disagreements never reach the model": "TestReconcile_TrivialDisputesAreResolvedWithoutAskingTheModel", "Trivial disagreements never reach the model": "TestReconcile_TrivialDisputesAreResolvedWithoutAskingTheModel",