Compare commits
29
Commits
+8
-2
@@ -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:*
|
||||
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).
|
||||
- Per-transcript deletion is not built; account deletion cascades. A private-content store
|
||||
without a delete button is a gap worth a ticket, not a silent omission.
|
||||
- ~~Per-transcript deletion is not built; account deletion cascades.~~ **Built 2026-10-03
|
||||
(#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
|
||||
surface, one image change. Removing the upload route removes the feature; `transcripts`, `videos`
|
||||
|
||||
@@ -44,7 +44,7 @@ func (f *fakeUploadsStore) SaveProgress(context.Context, string, string, []domai
|
||||
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) {
|
||||
f.sweeps++
|
||||
|
||||
@@ -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.
|
||||
- **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
|
||||
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
|
||||
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
|
||||
|
||||
@@ -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
|
||||
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
|
||||
When too many recordings are already waiting
|
||||
Then the upload is answered as busy, its row is failed honestly and its audio is deleted
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -37,6 +37,21 @@ type Client struct {
|
||||
temp float64
|
||||
httpClient *http.Client
|
||||
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
|
||||
@@ -101,6 +116,7 @@ func New(baseURL, apiKey, model string, timeout time.Duration, opts ...Option) *
|
||||
maxTokens: defaultMaxTokens,
|
||||
temp: defaultTemperature,
|
||||
httpClient: &http.Client{Timeout: timeout},
|
||||
maxBody: defaultMaxResponseBytes,
|
||||
}
|
||||
for _, opt := range opts {
|
||||
opt(c)
|
||||
@@ -184,12 +200,18 @@ func (c *Client) Complete(ctx context.Context, system, user string) (string, err
|
||||
}
|
||||
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 {
|
||||
return "", fmt.Errorf("read response: %w", err)
|
||||
}
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return "", &HTTPError{Code: resp.StatusCode, Body: string(out)}
|
||||
if int64(len(out)) > c.maxBody {
|
||||
return "", fmt.Errorf("LLM response exceeds %d bytes", c.maxBody)
|
||||
}
|
||||
|
||||
var cr chatResponse
|
||||
|
||||
@@ -17,6 +17,10 @@ import (
|
||||
// nothing about whether someone else's upload exists (ADR-033).
|
||||
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
|
||||
// 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
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
// Heartbeat stamps the upload as alive now. It does not touch progress_at: alive says
|
||||
// the service is still reporting, not that work is moving (ADR-035).
|
||||
func (s *Store) Heartbeat(ctx context.Context, userID, id string) error {
|
||||
// Heartbeat stamps the upload as alive at the given time. It does not touch progress_at:
|
||||
// alive says the service is still reporting, not that work is moving (ADR-035). The time
|
||||
// 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",
|
||||
`UPDATE uploaded_transcripts SET alive_at = NOW()
|
||||
WHERE id = $1 AND user_id = $2`)
|
||||
`UPDATE uploaded_transcripts SET alive_at = $3
|
||||
WHERE id = $1 AND user_id = $2`, at)
|
||||
}
|
||||
|
||||
// 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))
|
||||
}
|
||||
|
||||
// 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,
|
||||
// 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).
|
||||
|
||||
@@ -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()
|
||||
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))
|
||||
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)
|
||||
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")
|
||||
}
|
||||
|
||||
@@ -99,7 +99,7 @@ func TestUploadProgress_ByAnotherUserChangesNothing(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
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, errBeat, store.ErrUploadNotFound)
|
||||
|
||||
@@ -37,12 +37,13 @@ type Reporter interface {
|
||||
// is for people, so a failed write is logged and never stops the job.
|
||||
type Sink interface {
|
||||
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 (
|
||||
defaultMinSave = 2 * time.Second
|
||||
logAdvanceEach = time.Minute
|
||||
defaultMinSave = 2 * time.Second
|
||||
defaultSaveTimeout = 5 * time.Second
|
||||
logAdvanceEach = time.Minute
|
||||
)
|
||||
|
||||
// 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.
|
||||
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.
|
||||
type Tracker struct {
|
||||
ctx context.Context
|
||||
@@ -63,6 +72,8 @@ type Tracker struct {
|
||||
now func() time.Time
|
||||
minSave time.Duration
|
||||
|
||||
saveTimeout time.Duration
|
||||
|
||||
mu sync.Mutex
|
||||
steps []domain.ProgressStep
|
||||
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
|
||||
// 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 {
|
||||
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 {
|
||||
o(t)
|
||||
}
|
||||
@@ -212,7 +223,10 @@ func (t *Tracker) Heartbeat(ctx context.Context, beatEvery, stillEvery time.Dura
|
||||
case <-ctx.Done():
|
||||
return
|
||||
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)
|
||||
}
|
||||
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.
|
||||
func (t *Tracker) save() {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -17,11 +17,12 @@ import (
|
||||
|
||||
// fakeSink records every saved snapshot and every heartbeat.
|
||||
type fakeSink struct {
|
||||
mu sync.Mutex
|
||||
saves [][]domain.ProgressStep
|
||||
at []time.Time
|
||||
beats int
|
||||
err error
|
||||
mu sync.Mutex
|
||||
saves [][]domain.ProgressStep
|
||||
at []time.Time
|
||||
beats int
|
||||
beatAt []time.Time
|
||||
err 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
|
||||
}
|
||||
|
||||
func (f *fakeSink) Beat(context.Context) error {
|
||||
func (f *fakeSink) Beat(_ context.Context, at time.Time) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.beats++
|
||||
f.beatAt = append(f.beatAt, at)
|
||||
return f.err
|
||||
}
|
||||
|
||||
@@ -297,7 +299,7 @@ func (r *retainingSink) SaveProgress(_ context.Context, steps []domain.ProgressS
|
||||
r.got = append(r.got, steps)
|
||||
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) {
|
||||
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.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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -47,7 +47,7 @@ type WorkerStore interface {
|
||||
CompleteUpload(ctx context.Context, userID, id string, t domain.Transcription) 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
|
||||
Heartbeat(ctx context.Context, userID, id string) error
|
||||
Heartbeat(ctx context.Context, userID, id string, at time.Time) error
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
|
||||
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
|
||||
// not a second deployable: it shares the web process's lifetime, which is why it
|
||||
|
||||
@@ -70,7 +70,7 @@ func (s *wStore) SaveProgress(_ context.Context, _, _ string, steps []domain.Pro
|
||||
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)
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
@@ -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
|
||||
// health probes carry no headers: safe methods must never be refused on origin grounds.
|
||||
func TestCrossOrigin_ACrossSiteGETIsNotRefused(t *testing.T) {
|
||||
|
||||
@@ -59,6 +59,9 @@ type Store interface {
|
||||
// non-owner's id is ErrUploadNotFound.
|
||||
GetUpload(ctx context.Context, userID, id 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.
|
||||
// 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("GET /uploads/{id}", a.handleUploadDetail)
|
||||
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
|
||||
// when one is wired.
|
||||
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))
|
||||
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).
|
||||
|
||||
@@ -66,7 +66,7 @@ func TestParseYouTubeVideoID(t *testing.T) {
|
||||
func TestListPageShowsPasteFormOnlyWhenConnected(t *testing.T) {
|
||||
render := func(connected bool) string {
|
||||
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)
|
||||
}
|
||||
return buf.String()
|
||||
|
||||
@@ -18,6 +18,7 @@ import (
|
||||
|
||||
"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/config"
|
||||
"git.d-ma.be/mathias/tapir/internal/domain"
|
||||
"git.d-ma.be/mathias/tapir/internal/errkind"
|
||||
"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()))
|
||||
}
|
||||
|
||||
// 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) {
|
||||
userID, ok := a.currentUserID(w, r)
|
||||
if !ok {
|
||||
@@ -243,6 +270,36 @@ func titleFromFilename(name string) string {
|
||||
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
|
||||
// recording that is shorter than its file claimed, so a record that has one is shown open.
|
||||
func hasNote(steps []domain.ProgressStep) bool {
|
||||
|
||||
@@ -66,6 +66,12 @@ templ uploadStatus(u domain.UploadedTranscript, chatEnabled bool, now time.Time)
|
||||
Transcribed { uploads.FormatDuration(time.Duration(u.AudioSeconds * float64(time.Second))) }
|
||||
of audio by <span class="chip">{ u.Endpoint }</span>.
|
||||
</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)
|
||||
if len(u.Progress) > 0 {
|
||||
<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>
|
||||
<pre class="transcript">{ u.Content }</pre>
|
||||
</details>
|
||||
@deleteRecording(u.ID)
|
||||
</div>
|
||||
case domain.UploadFailed:
|
||||
<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)
|
||||
</details>
|
||||
}
|
||||
@deleteRecording(u.ID)
|
||||
</div>
|
||||
default:
|
||||
<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:
|
||||
// the same chat panel as a video's (ADR-027), scoped to this recording.
|
||||
templ UploadChatPage(u domain.UploadedTranscript, v chatView) {
|
||||
|
||||
@@ -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"`))
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
+410
-335
File diff suppressed because it is too large
Load Diff
@@ -107,9 +107,12 @@ templ flashBanner(code string) {
|
||||
// #summary-list region; a non-HTMX request renders the whole page. flash carries
|
||||
// a one-shot notification (e.g. "connected", "registered") surfaced on arrival
|
||||
// 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") {
|
||||
@flashBanner(flash)
|
||||
if uploadsEnabled {
|
||||
<p class="muted"><a href="/uploads">Recordings</a>: upload audio and get a private transcript.</p>
|
||||
}
|
||||
if hasConnected {
|
||||
@pasteForm()
|
||||
}
|
||||
|
||||
+358
-348
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",
|
||||
"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
|
||||
"Two engines transcribe the audio and their disagreements are reconciled": "TestEnsemble_BothEnginesRunInOrderAndTheirDisagreementIsReconciled",
|
||||
"Trivial disagreements never reach the model": "TestReconcile_TrivialDisputesAreResolvedWithoutAskingTheModel",
|
||||
|
||||
Reference in New Issue
Block a user