Compare commits

...
11 Commits
Author SHA1 Message Date
mathias b9643f5fb7 Merge pull request 'feat(uploads): choose the language of a recording; Swedish default, Other = auto (#61 interim)' (#62) from sprint/language into main
CI / Deploy via GitOps (push) Successful in 5s
CI / Lint / Test / Vet (push) Successful in 34s
CI / Build & Import (push) Successful in 14s
2026-10-05 04:59:03 +00:00
mathiasandClaude Sonnet 5.5 f704258fdf test(transcribe): the wrapper does not restart a cancelled job; PlanFor of an empty or unknown language is pinned (#61)
CI / Lint / Test / Vet (pull_request) Successful in 30s
CI / Build & Import (pull_request) Skipped
CI / Deploy via GitOps (pull_request) Skipped
Low findings of the PR #62 re-review: the WithTimeout test's comment claimed cancellation
travels on the context but only the language was asserted (the WithoutCancel mutant survived),
and PlanFor('') and PlanFor(unknown) were untested.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-04 22:48:29 +02:00
mathiasandClaude Sonnet 5.5 e3288d0654 feat(web): the language is named on the recording's page in every state; the feature says what Other really does (#61)
CI / Build & Import (pull_request) Skipped
CI / Deploy via GitOps (pull_request) Skipped
CI / Lint / Test / Vet (pull_request) Successful in 37s
The label moves from the done fragment to the page header, so queued, transcribing and
failed recordings show it too. The scenario no longer implies a general engine exists. Found
by the PR #62 review.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-04 22:41:42 +02:00
mathiasandClaude Sonnet 5.5 e0c53f8719 feat(uploads): an auto-language job is planned without the engines it never runs (#61)
PlanFor(lang) on the Ensemble and a LanguagePlanner the worker prefers: the page lists
the other engine (and the fallback if configured), not the base engine, the comparison and
the decisions as coming. Found by the PR #62 review.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-04 22:41:21 +02:00
mathiasandClaude Sonnet 5.5 0828f0debf test(web): the language refusal asserts its message, not a word the form always contains (#61)
Found by the PR #62 review: the message could be replaced with 'x' and the test stayed green.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-04 22:40:47 +02:00
mathiasandClaude Sonnet 5.5 7282893877 test(transcribe): WithTimeout hands the engine the job's language, as on the production path (#61)
The wrapper sits in front of both local engines; a fresh context there would send an auto
upload as language=sv and un-cancel the job, and no test noticed. Found by the PR #62 review.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-04 22:40:35 +02:00
mathiasandClaude Sonnet 5.5 cc6fe7083d refactor(web): drop a length check the closed-set parse already makes (#61)
CI / Lint / Test / Vet (pull_request) Successful in 36s
CI / Build & Import (pull_request) Skipped
CI / Deploy via GitOps (pull_request) Skipped
The read is capped at 17 bytes and no valid language is longer than 4, so the parse
refuses anything the length check would; the mutant that removed it survived.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-04 22:31:03 +02:00
mathiasandClaude Sonnet 5.5 0338b18a17 feat(uploads): choose the language of a recording; record it, show it, carry it to the engines (#61)
Form field (Swedish preselected, Other = auto) read before the file part and refused
before anything is spooled if outside the closed set; stored on the row (migration 021,
CHECK constraint, default sv for history); carried on the Job and the worker's context;
shown on the status page. Feature scenarios and the coverage map are updated, and
ADR-033/034 amended.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-04 22:29:58 +02:00
mathiasandClaude Sonnet 5.5 ed9d76b0cf feat(transcribe): a non-Swedish upload skips the Swedish-only base engine and is not reconciled (#61)
The other engine transcribes alone, then the cloud fallback if it fails and the job is
still live. A cancelled job never reaches the fallback (mutation-checked).

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-04 22:26:37 +02:00
mathiasandClaude Sonnet 5.5 0b807c6bcd feat(transcribe): the gateway sends language=sv, omits it for auto, and refuses anything else (#61)
The request used to hard-code language=sv for every upload, so a recording in another
language was decoded as Swedish. The language now comes from the context; no language in
the context is still sv.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-04 22:26:13 +02:00
mathiasandClaude Sonnet 5.5 8e4872d2e7 feat(domain): Language, a closed set carried to the engines on the context (#61)
sv and auto; an empty form value is sv, anything else is refused. LanguageFrom defaults to
sv so a caller that sets nothing behaves as before.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-04 22:25:51 +02:00
35 changed files with 1100 additions and 401 deletions
+10 -1
View File
@@ -1434,7 +1434,12 @@ an hour of Swedish *was* an unknown here; it is now measured, see decision 8.)
empty disables), recorded per row, and called out in Open items rather than defaulted empty disables), recorded per row, and called out in Open items rather than defaulted
silently. silently.
- Swedish only (`language=sv`). Language detection and routing (#28 decision 2) are not - Swedish only (`language=sv`). Language detection and routing (#28 decision 2) are not
touched. touched. **Amended 2026-10-04 (#61, interim):** the language is now a per-upload choice on
the form, `sv` (preselected, and what an upload that names none gets) or `auto` ("Other":
`language` is omitted so the engine detects it), a closed set held by the form, the request
code and a CHECK constraint on `uploaded_transcripts.language` (migration 021). Automatic
detection and a general-purpose engine for other languages are still not built (#61); an
`auto` upload goes to the same configured engine, which is not tuned for it.
- `-H 'Expect:'`, multipart field `file` — encoded in the adapter and its tests, not rediscovered. - `-H 'Expect:'`, multipart field `file` — encoded in the adapter and its tests, not rediscovered.
7. **Jobs run in-process, one at a time (ADR-018 stands).** Transcription is minutes (~7.5 min 7. **Jobs run in-process, one at a time (ADR-018 stands).** Transcription is minutes (~7.5 min
@@ -1645,6 +1650,10 @@ on the slow numbers and is reopened: 31B scored 97% against 92%.
job timeout rises from 30 to 60 minutes: ~4 min of kb-whisper and ~40 s of pianissimo per hour of job timeout rises from 30 to 60 minutes: ~4 min of kb-whisper and ~40 s of pianissimo per hour of
audio, plus the adjudication. audio, plus the adjudication.
**Amended 2026-10-04 (#61, interim):** reconcile is for Swedish. A recording whose language is
`auto` is never sent to the base engine (it is Swedish-only) and is not reconciled: the other
engine transcribes alone, then the cloud fallback if that fails and the job is live, and the
page does not claim a second opinion.
8. **Not decided here.** A cloud adjudicator (rejected: private content); tuning the prompt; making 8. **Not decided here.** A cloud adjudicator (rejected: private content); tuning the prompt; making
`gemma4-31b` the default; a "reconciling" progress state (the page says "transcribing" for the whole `gemma4-31b` the default; a "reconciling" progress state (the page says "transcribing" for the whole
job); feeding the uncertain spots to chat; whether pianissimo should also be a middle tier in the job); feeding the uncertain spots to chat; whether pianissimo should also be a middle tier in the
+1 -1
View File
@@ -25,7 +25,7 @@ type fakeUploadsStore struct {
failedKinds []domain.UploadErrorKind failedKinds []domain.UploadErrorKind
} }
func (f *fakeUploadsStore) CreateUpload(context.Context, string, string, bool) (string, error) { func (f *fakeUploadsStore) CreateUpload(context.Context, string, string, domain.Language, bool) (string, error) {
return "id", nil return "id", nil
} }
+17
View File
@@ -274,3 +274,20 @@ Feature: Upload an audio recording, get a private Swedish transcript, discuss it
Scenario: Input that cannot be right is a client error Scenario: Input that cannot be right is a client error
When the id is not an id, or a header or title holds control bytes When the id is not an id, or a header or title holds control bytes
Then the answer is a 4xx or the title is cleaned, never a server error Then the answer is a 4xx or the title is cleaned, never a server error
Scenario: I choose the language of a recording
When I upload a recording and choose Swedish or Other on the form
Then the recording is transcribed as that language and its page says which
Scenario: Swedish stays the default
When an upload names no language
Then it is transcribed as Swedish, as before the choice existed
Scenario: A language outside the list is refused
When an upload names a language the form does not offer
Then it is answered 400 before anything is stored or sent to an engine
Scenario: A recording in another language never reaches the Swedish-only engine
When I choose Other for a recording
Then the language is left for the engine to detect and the Swedish-only base engine and the reconcile are not used
And the engine that does run is the configured one, which is not tuned for other languages
@@ -0,0 +1,2 @@
ALTER TABLE uploaded_transcripts
DROP COLUMN language;
@@ -0,0 +1,13 @@
-- Migration 021: the language an upload is transcribed as (tapir#61).
--
-- 'sv' Swedish: the engines are sent language=sv. Every row before this migration was
-- transcribed that way (the request hard-coded it), so the default is true history.
-- 'auto' Other: the engine detects the language and the Swedish-only engine is not used.
--
-- A closed set held by the table as well as by the form and the request code, so a second
-- writer or a bug cannot store a value the engines are never meant to receive. The column
-- sits on the already-FORCE'd, RLS-scoped uploaded_transcripts, so the exact-set isolation
-- test is unaffected.
ALTER TABLE uploaded_transcripts
ADD COLUMN language TEXT NOT NULL DEFAULT 'sv'
CONSTRAINT uploaded_transcripts_language_check CHECK (language IN ('sv', 'auto'));
@@ -31,12 +31,12 @@ var ErrUploadInProgress = errors.New("store: uploaded transcript is still being
// CreateUpload records a new upload in the queued state and returns its id. // CreateUpload records a new upload in the queued state and returns its id.
// hadVideo records that the file carried a video track (only its audio is used); // hadVideo records that the file carried a video track (only its audio is used);
// it is kept for the life of the row so the page can say so. // it is kept for the life of the row so the page can say so.
func (s *Store) CreateUpload(ctx context.Context, userID, title string, hadVideo bool) (string, error) { func (s *Store) CreateUpload(ctx context.Context, userID, title string, lang domain.Language, hadVideo bool) (string, error) {
var id string var id string
if err := s.withUser(ctx, userID, func(tx pgx.Tx) error { if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
return tx.QueryRow(ctx, return tx.QueryRow(ctx,
`INSERT INTO uploaded_transcripts (user_id, title, had_video) VALUES ($1, $2, $3) RETURNING id`, `INSERT INTO uploaded_transcripts (user_id, title, language, had_video) VALUES ($1, $2, $3, $4) RETURNING id`,
userID, title, hadVideo).Scan(&id) userID, title, string(lang), hadVideo).Scan(&id)
}); err != nil { }); err != nil {
return "", fmt.Errorf("store: create upload: %w", err) return "", fmt.Errorf("store: create upload: %w", err)
} }
@@ -221,7 +221,7 @@ func (s *Store) FailInterruptedUploads(ctx context.Context) (int, error) {
return total, nil return total, nil
} }
const uploadSelect = `SELECT id, user_id, title, status, content, segments, endpoint, const uploadSelect = `SELECT id, user_id, title, language, status, content, segments, endpoint,
audio_seconds, error_kind, had_video, reconciliation, progress, progress_at, alive_at, audio_seconds, error_kind, had_video, reconciliation, progress, progress_at, alive_at,
created_at, finished_at FROM uploaded_transcripts` created_at, finished_at FROM uploaded_transcripts`
@@ -234,11 +234,13 @@ func scanUpload(row pgx.Row, u *domain.UploadedTranscript, withText bool) error
seconds *float64 seconds *float64
finished, progressAt, aliveAt *time.Time finished, progressAt, aliveAt *time.Time
) )
if err := row.Scan(&u.ID, &u.UserID, &u.Title, &status, &content, &segments, &endpoint, var language string
if err := row.Scan(&u.ID, &u.UserID, &u.Title, &language, &status, &content, &segments, &endpoint,
&seconds, &errKind, &u.HadVideo, &reconciliation, &progress, &progressAt, &aliveAt, &seconds, &errKind, &u.HadVideo, &reconciliation, &progress, &progressAt, &aliveAt,
&u.CreatedAt, &finished); err != nil { &u.CreatedAt, &finished); err != nil {
return err return err
} }
u.Language = domain.Language(language)
u.Status = domain.UploadStatus(deref(status)) u.Status = domain.UploadStatus(deref(status))
u.Endpoint = deref(endpoint) u.Endpoint = deref(endpoint)
u.ErrorKind = domain.UploadErrorKind(deref(errKind)) u.ErrorKind = domain.UploadErrorKind(deref(errKind))
@@ -42,7 +42,7 @@ func TestDeleteUpload_RefusesAnUploadThatIsStillQueuedOrTranscribingAndKeepsTheR
s := newStore(t) s := newStore(t)
ctx := context.Background() ctx := context.Background()
queued := startedUpload(t, s) queued := startedUpload(t, s)
running, err := s.CreateUpload(ctx, userA, "running", false) running, err := s.CreateUpload(ctx, userA, "running", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
require.NoError(t, s.StartTranscribing(ctx, userA, running)) require.NoError(t, s.StartTranscribing(ctx, userA, running))
@@ -34,7 +34,7 @@ func TestFailUpload_StillFailsAQueuedOrTranscribingUpload(t *testing.T) {
s := newStore(t) s := newStore(t)
ctx := context.Background() ctx := context.Background()
queued := startedUpload(t, s) queued := startedUpload(t, s)
running, err := s.CreateUpload(ctx, userA, "r", false) running, err := s.CreateUpload(ctx, userA, "r", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
require.NoError(t, s.StartTranscribing(ctx, userA, running)) require.NoError(t, s.StartTranscribing(ctx, userA, running))
@@ -20,7 +20,7 @@ import (
func startedUpload(t *testing.T, s *store.Store) string { func startedUpload(t *testing.T, s *store.Store) string {
t.Helper() t.Helper()
seedUsers(t, userA, userB) seedUsers(t, userA, userB)
id, err := s.CreateUpload(context.Background(), userA, "m", false) id, err := s.CreateUpload(context.Background(), userA, "m", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
return id return id
} }
@@ -33,7 +33,7 @@ func TestCreateUpload_StartsQueuedAndIsOwnedByTheCreator(t *testing.T) {
seedUsers(t, userA) seedUsers(t, userA)
ctx := context.Background() ctx := context.Background()
id, err := s.CreateUpload(ctx, userA, "Teams export", false) id, err := s.CreateUpload(ctx, userA, "Teams export", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
got, err := s.GetUpload(ctx, userA, id) got, err := s.GetUpload(ctx, userA, id)
@@ -50,7 +50,7 @@ func TestUploadLifecycle_CompletedKeepsTextSegmentsEndpointAndDuration(t *testin
s := newStore(t) s := newStore(t)
seedUsers(t, userA) seedUsers(t, userA)
ctx := context.Background() ctx := context.Background()
id, err := s.CreateUpload(ctx, userA, "m", false) id, err := s.CreateUpload(ctx, userA, "m", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
require.NoError(t, s.StartTranscribing(ctx, userA, id)) require.NoError(t, s.StartTranscribing(ctx, userA, id))
@@ -84,7 +84,7 @@ func TestUploadLifecycle_FailedRecordsTypedKindAndNoContent(t *testing.T) {
s := newStore(t) s := newStore(t)
seedUsers(t, userA) seedUsers(t, userA)
ctx := context.Background() ctx := context.Background()
id, err := s.CreateUpload(ctx, userA, "m", false) id, err := s.CreateUpload(ctx, userA, "m", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
require.NoError(t, s.StartTranscribing(ctx, userA, id)) require.NoError(t, s.StartTranscribing(ctx, userA, id))
@@ -102,7 +102,7 @@ func TestGetUpload_AnotherUsersIDIsNotFound(t *testing.T) {
s := newStore(t) s := newStore(t)
seedUsers(t, userA, userB) seedUsers(t, userA, userB)
ctx := context.Background() ctx := context.Background()
id, err := s.CreateUpload(ctx, userA, "A's meeting", false) id, err := s.CreateUpload(ctx, userA, "A's meeting", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
_, err = s.GetUpload(ctx, userB, id) _, err = s.GetUpload(ctx, userB, id)
@@ -115,7 +115,7 @@ func TestUploadWrites_ByAnotherUserChangeNothing(t *testing.T) {
s := newStore(t) s := newStore(t)
seedUsers(t, userA, userB) seedUsers(t, userA, userB)
ctx := context.Background() ctx := context.Background()
id, err := s.CreateUpload(ctx, userA, "A's meeting", false) id, err := s.CreateUpload(ctx, userA, "A's meeting", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
assert.ErrorIs(t, s.StartTranscribing(ctx, userB, id), store.ErrUploadNotFound) assert.ErrorIs(t, s.StartTranscribing(ctx, userB, id), store.ErrUploadNotFound)
@@ -132,11 +132,11 @@ func TestListUploads_ReturnsOnlyTheCallersNewestFirstWithoutTranscriptText(t *te
s := newStore(t) s := newStore(t)
seedUsers(t, userA, userB) seedUsers(t, userA, userB)
ctx := context.Background() ctx := context.Background()
first, err := s.CreateUpload(ctx, userA, "first", false) first, err := s.CreateUpload(ctx, userA, "first", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
second, err := s.CreateUpload(ctx, userA, "second", false) second, err := s.CreateUpload(ctx, userA, "second", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
_, err = s.CreateUpload(ctx, userB, "B's", false) _, err = s.CreateUpload(ctx, userB, "B's", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
require.NoError(t, s.StartTranscribing(ctx, userA, first)) require.NoError(t, s.StartTranscribing(ctx, userA, first))
require.NoError(t, s.CompleteUpload(ctx, userA, first, domain.Transcription{Text: "long text body", Endpoint: "e"})) require.NoError(t, s.CompleteUpload(ctx, userA, first, domain.Transcription{Text: "long text body", Endpoint: "e"}))
@@ -160,12 +160,12 @@ func TestFailInterruptedUploads_FailsOnlyUnfinishedRowsForEveryUser(t *testing.T
_, err := super.Exec(ctx, `INSERT INTO user_identities (dex_subject, user_id) VALUES ($1, $2)`, subject, uid) _, err := super.Exec(ctx, `INSERT INTO user_identities (dex_subject, user_id) VALUES ($1, $2)`, subject, uid)
require.NoError(t, err) require.NoError(t, err)
} }
queued, err := s.CreateUpload(ctx, userA, "queued", false) queued, err := s.CreateUpload(ctx, userA, "queued", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
running, err := s.CreateUpload(ctx, userB, "running", false) running, err := s.CreateUpload(ctx, userB, "running", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
require.NoError(t, s.StartTranscribing(ctx, userB, running)) require.NoError(t, s.StartTranscribing(ctx, userB, running))
done, err := s.CreateUpload(ctx, userA, "done", false) done, err := s.CreateUpload(ctx, userA, "done", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
require.NoError(t, s.StartTranscribing(ctx, userA, done)) require.NoError(t, s.StartTranscribing(ctx, userA, done))
require.NoError(t, s.CompleteUpload(ctx, userA, done, domain.Transcription{Text: "kept", Endpoint: "e"})) require.NoError(t, s.CompleteUpload(ctx, userA, done, domain.Transcription{Text: "kept", Endpoint: "e"}))
@@ -191,9 +191,9 @@ func TestCreateUpload_RecordsWhetherTheFileHadAVideoTrack(t *testing.T) {
s := newStore(t) s := newStore(t)
seedUsers(t, userA) seedUsers(t, userA)
ctx := context.Background() ctx := context.Background()
withVideo, err := s.CreateUpload(ctx, userA, "camera off", true) withVideo, err := s.CreateUpload(ctx, userA, "camera off", domain.LangSwedish, true)
require.NoError(t, err) require.NoError(t, err)
audioOnly, err := s.CreateUpload(ctx, userA, "voice only", false) audioOnly, err := s.CreateUpload(ctx, userA, "voice only", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
gotVideo, err := s.GetUpload(ctx, userA, withVideo) gotVideo, err := s.GetUpload(ctx, userA, withVideo)
@@ -216,9 +216,9 @@ func TestUploadHadVideo_SurvivesCompletionAndFailure(t *testing.T) {
s := newStore(t) s := newStore(t)
seedUsers(t, userA) seedUsers(t, userA)
ctx := context.Background() ctx := context.Background()
done, err := s.CreateUpload(ctx, userA, "d", true) done, err := s.CreateUpload(ctx, userA, "d", domain.LangSwedish, true)
require.NoError(t, err) require.NoError(t, err)
failed, err := s.CreateUpload(ctx, userA, "f", true) failed, err := s.CreateUpload(ctx, userA, "f", domain.LangSwedish, true)
require.NoError(t, err) require.NoError(t, err)
require.NoError(t, s.StartTranscribing(ctx, userA, done)) require.NoError(t, s.StartTranscribing(ctx, userA, done))
require.NoError(t, s.CompleteUpload(ctx, userA, done, domain.Transcription{Text: "x", Endpoint: "e"})) require.NoError(t, s.CompleteUpload(ctx, userA, done, domain.Transcription{Text: "x", Endpoint: "e"}))
@@ -235,7 +235,7 @@ func TestUploadLifecycle_ReconciliationRoundTripsWithItsContestedSpots(t *testin
s := newStore(t) s := newStore(t)
seedUsers(t, userA) seedUsers(t, userA)
ctx := context.Background() ctx := context.Background()
id, err := s.CreateUpload(ctx, userA, "m", false) id, err := s.CreateUpload(ctx, userA, "m", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
require.NoError(t, s.StartTranscribing(ctx, userA, id)) require.NoError(t, s.StartTranscribing(ctx, userA, id))
want := &domain.Reconciliation{ want := &domain.Reconciliation{
@@ -258,7 +258,7 @@ func TestUploadLifecycle_ASingleEngineTranscriptHasNoReconciliation(t *testing.T
s := newStore(t) s := newStore(t)
seedUsers(t, userA) seedUsers(t, userA)
ctx := context.Background() ctx := context.Background()
id, err := s.CreateUpload(ctx, userA, "m", false) id, err := s.CreateUpload(ctx, userA, "m", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
require.NoError(t, s.StartTranscribing(ctx, userA, id)) require.NoError(t, s.StartTranscribing(ctx, userA, id))
require.NoError(t, s.CompleteUpload(ctx, userA, id, domain.Transcription{Text: "x", Endpoint: "e"})) require.NoError(t, s.CompleteUpload(ctx, userA, id, domain.Transcription{Text: "x", Endpoint: "e"}))
@@ -273,7 +273,7 @@ func TestUploadLifecycle_ASkippedReconciliationKeepsItsReason(t *testing.T) {
s := newStore(t) s := newStore(t)
seedUsers(t, userA) seedUsers(t, userA)
ctx := context.Background() ctx := context.Background()
id, err := s.CreateUpload(ctx, userA, "m", false) id, err := s.CreateUpload(ctx, userA, "m", domain.LangSwedish, false)
require.NoError(t, err) require.NoError(t, err)
require.NoError(t, s.StartTranscribing(ctx, userA, id)) require.NoError(t, s.StartTranscribing(ctx, userA, id))
rec := &domain.Reconciliation{Status: domain.ReconcileSkipped, Reason: "the model failed 3 times in a row: gateway down", Errors: 3} rec := &domain.Reconciliation{Status: domain.ReconcileSkipped, Reason: "the model failed 3 times in a row: gateway down", Errors: 3}
@@ -284,3 +284,55 @@ func TestUploadLifecycle_ASkippedReconciliationKeepsItsReason(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, rec, got.Reconciliation) assert.Equal(t, rec, got.Reconciliation)
} }
func TestCreateUpload_RecordsTheLanguageTheUserChoseAndTheListCarriesIt(t *testing.T) {
s := newStore(t)
seedUsers(t, userA)
ctx := context.Background()
swedish, err := s.CreateUpload(ctx, userA, "sv", domain.LangSwedish, false)
require.NoError(t, err)
auto, err := s.CreateUpload(ctx, userA, "auto", domain.LangAuto, false)
require.NoError(t, err)
gotSv, err := s.GetUpload(ctx, userA, swedish)
require.NoError(t, err)
gotAuto, err := s.GetUpload(ctx, userA, auto)
require.NoError(t, err)
listed, err := s.ListUploads(ctx, userA)
require.NoError(t, err)
assert.Equal(t, domain.LangSwedish, gotSv.Language)
assert.Equal(t, domain.LangAuto, gotAuto.Language)
byID := map[string]domain.Language{}
for _, u := range listed {
byID[u.ID] = u.Language
}
assert.Equal(t, map[string]domain.Language{swedish: domain.LangSwedish, auto: domain.LangAuto}, byID)
}
// The closed set is also held by the table, so a bug or a second writer cannot store a
// value the engines are never meant to receive (tapir#61).
func TestUploadedTranscripts_TheTableRefusesALanguageOutsideTheClosedSet(t *testing.T) {
newStore(t)
seedUsers(t, userA)
_, err := rawPool(t).Exec(context.Background(),
`INSERT INTO uploaded_transcripts (user_id, title, language) VALUES ($1, 'x', 'en')`, userA)
require.Error(t, err)
assert.Contains(t, err.Error(), "language", "the CHECK constraint names the column")
}
// A row from before the column existed was transcribed as Swedish; the default says so.
func TestUploadedTranscripts_ARowInsertedWithoutALanguageIsSwedish(t *testing.T) {
s := newStore(t)
seedUsers(t, userA)
var id string
require.NoError(t, rawPool(t).QueryRow(context.Background(),
`INSERT INTO uploaded_transcripts (user_id, title) VALUES ($1, 'old') RETURNING id`, userA).Scan(&id))
got, err := s.GetUpload(context.Background(), userA, id)
require.NoError(t, err)
assert.Equal(t, domain.LangSwedish, got.Language)
}
+43 -8
View File
@@ -32,14 +32,24 @@ func NewEnsemble(base, alt, fallback Backend, adj reconcile.Adjudicator, opts re
return &Ensemble{base: base, alt: alt, fallback: fallback, adj: adj, opts: opts} return &Ensemble{base: base, alt: alt, fallback: fallback, adj: adj, opts: opts}
} }
// Plan implements uploads.Planner: the phases this ensemble reports, so the page can // Plan implements uploads.Planner: the phases of a Swedish job, the two-engine one. The
// show them before they run. The fallback is a phase only when one is configured. // fallback is a phase only when one is configured.
func (e *Ensemble) Plan() []progress.Planned { func (e *Ensemble) Plan() []progress.Planned { return e.PlanFor(domain.LangSwedish) }
plan := []progress.Planned{
{Key: "engine_a", Label: "Transcribing with " + e.base.Name()}, // PlanFor implements uploads.LanguagePlanner. A recording that is not Swedish never runs the
{Key: "engine_b", Label: "Transcribing with " + e.alt.Name()}, // base engine, the comparison or the decisions (transcribeOther), so the page lists only
{Key: "compare", Label: "Comparing the two transcripts"}, // what can run: the other engine, then the fallback if one is configured.
{Key: "decide", Label: "Deciding the differences"}, func (e *Ensemble) PlanFor(lang domain.Language) []progress.Planned {
var plan []progress.Planned
if lang == domain.LangSwedish || lang == "" {
plan = []progress.Planned{
{Key: "engine_a", Label: "Transcribing with " + e.base.Name()},
{Key: "engine_b", Label: "Transcribing with " + e.alt.Name()},
{Key: "compare", Label: "Comparing the two transcripts"},
{Key: "decide", Label: "Deciding the differences"},
}
} else {
plan = []progress.Planned{{Key: "engine_b", Label: "Transcribing with " + e.alt.Name()}}
} }
if e.fallback != nil { if e.fallback != nil {
plan = append(plan, progress.Planned{Key: "fallback", Label: "Cloud fallback (" + e.fallback.Name() + ")"}) plan = append(plan, progress.Planned{Key: "fallback", Label: "Cloud fallback (" + e.fallback.Name() + ")"})
@@ -51,6 +61,9 @@ func (e *Ensemble) Plan() []progress.Planned {
// progress.Reporter in ctx (a no-op when nobody listens). // progress.Reporter in ctx (a no-op when nobody listens).
func (e *Ensemble) Transcribe(ctx context.Context, wavPath string) (domain.Transcription, error) { func (e *Ensemble) Transcribe(ctx context.Context, wavPath string) (domain.Transcription, error) {
rep := progress.From(ctx) rep := progress.From(ctx)
if domain.LanguageFrom(ctx) != domain.LangSwedish {
return e.transcribeOther(ctx, rep, wavPath)
}
b, berr := e.run(ctx, rep, "engine_a", e.base, wavPath) b, berr := e.run(ctx, rep, "engine_a", e.base, wavPath)
if err := ctx.Err(); err != nil { if err := ctx.Err(); err != nil {
return domain.Transcription{}, err return domain.Transcription{}, err
@@ -89,6 +102,28 @@ func (e *Ensemble) Transcribe(ctx context.Context, wavPath string) (domain.Trans
return e.single(f, e.fallback.Name(), "both local engines failed"), nil return e.single(f, e.fallback.Name(), "both local engines failed"), nil
} }
// transcribeOther handles a recording that is not Swedish (tapir#61). The base engine is
// Swedish-only, so it never receives the audio, and with one reading there is nothing to
// reconcile: the other engine transcribes alone, then the cloud fallback if it fails and the
// job is still live. The phases that did not run are shown as skipped by the tracker.
func (e *Ensemble) transcribeOther(ctx context.Context, rep progress.Reporter, wavPath string) (domain.Transcription, error) {
t, err := e.run(ctx, rep, "engine_b", e.alt, wavPath)
if err == nil {
t.Endpoint = e.alt.Name()
return t, nil
}
if e.fallback == nil || ctx.Err() != nil {
return domain.Transcription{}, fmt.Errorf("%s: %w", e.alt.Name(), err)
}
f, ferr := e.run(ctx, rep, "fallback", e.fallback, wavPath)
if ferr != nil {
return domain.Transcription{}, errors.Join(
fmt.Errorf("%s: %w", e.alt.Name(), err), fmt.Errorf("%s: %w", e.fallback.Name(), ferr))
}
f.Endpoint = e.fallback.Name()
return f, nil
}
// run is one engine call, reported as a phase. A job that was stopped is not a broken // run is one engine call, reported as a phase. A job that was stopped is not a broken
// engine, so a cancelled context is never reported as a failed phase. // engine, so a cancelled context is never reported as a failed phase.
func (e *Ensemble) run(ctx context.Context, rep progress.Reporter, key string, be Backend, wav string) (domain.Transcription, error) { func (e *Ensemble) run(ctx context.Context, rep progress.Reporter, key string, be Backend, wav string) (domain.Transcription, error) {
@@ -160,3 +160,25 @@ func TestEnsembleProgress_TheReconcileCountsAreNotedForTheOperator(t *testing.T)
assert.Equal(t, 0, n["unsure"]) assert.Equal(t, 0, n["unsure"])
assert.Equal(t, 0, n["errors"]) assert.Equal(t, 0, n["errors"])
} }
// An auto-language upload never runs the base engine, the comparison or the decisions, so the
// page must not list them as coming (tapir#61, PR #62 review).
func TestEnsemblePlanFor_AnAutoLanguageListsOnlyThePhasesThatCanRun(t *testing.T) {
with := newEnsemble("a", "a", nil, nil, true, &verdictAdj{}).e
without := newEnsemble("a", "a", nil, nil, false, &verdictAdj{}).e
keys := func(p []progress.Planned) []string {
var k []string
for _, s := range p {
k = append(k, s.Key)
}
return k
}
assert.Equal(t, []string{"engine_b", "fallback"}, keys(with.PlanFor(domain.LangAuto)))
assert.Equal(t, []string{"engine_b"}, keys(without.PlanFor(domain.LangAuto)), "no fallback configured: none promised")
assert.Equal(t, with.Plan(), with.PlanFor(domain.LangSwedish), "Swedish is the plan it always was")
assert.Contains(t, with.PlanFor(domain.LangAuto)[0].Label, "iguana/kb-whisper")
assert.Equal(t, with.Plan(), with.PlanFor(""), "a job queued before the choice existed is Swedish")
assert.Equal(t, with.PlanFor(domain.LangAuto), with.PlanFor(domain.Language("xx")),
"an unknown language is planned as it is run: not the Swedish phases")
}
@@ -204,3 +204,52 @@ func TestEnsemble_AShutdownDuringTheSecondEngineIsAnErrorNotASingleEngineSuccess
assert.ErrorIs(t, err, context.Canceled, "a cancelled job must not be recorded as 'kb-whisper failed'") assert.ErrorIs(t, err, context.Canceled, "a cancelled job must not be recorded as 'kb-whisper failed'")
assert.Empty(t, got.Text) assert.Empty(t, got.Text)
} }
// The base engine is Swedish-only. A recording that is not Swedish must never reach it, and
// there is no second opinion to reconcile: the other engine transcribes alone (tapir#61).
func autoCtx() context.Context { return domain.WithLanguage(context.Background(), domain.LangAuto) }
func TestEnsemble_AnAutoLanguageUploadNeverReachesTheBaseEngineAndIsNotReconciled(t *testing.T) {
adj := &verdictAdj{v: reconcile.VerdictAlt}
r := newEnsemble("base text", "alt text", nil, nil, true, adj)
got, err := r.e.Transcribe(autoCtx(), "/a.wav")
require.NoError(t, err)
assert.Equal(t, []string{"iguana/kb-whisper"}, *r.calls, "only the other engine is called")
assert.Zero(t, adj.calls, "nothing is adjudicated")
assert.Equal(t, "alt text", got.Text)
assert.Equal(t, "iguana/kb-whisper", got.Endpoint)
assert.Nil(t, got.Reconciliation, "reconciliation does not apply, so none is recorded")
}
func TestEnsemble_AnAutoLanguageUploadFallsBackWhenItsEngineFailsAndStillSkipsTheBaseEngine(t *testing.T) {
r := newEnsemble("base text", "alt text", nil, errors.New("engine down"), true, &verdictAdj{})
got, err := r.e.Transcribe(autoCtx(), "/a.wav")
require.NoError(t, err)
assert.Equal(t, []string{"iguana/kb-whisper", "berget/kb-whisper"}, *r.calls)
assert.Equal(t, "berget/kb-whisper", got.Endpoint)
}
func TestEnsemble_AnAutoLanguageUploadWithNoFallbackFailsWithoutTouchingTheBaseEngine(t *testing.T) {
r := newEnsemble("base text", "alt text", nil, errors.New("engine down"), false, &verdictAdj{})
_, err := r.e.Transcribe(autoCtx(), "/a.wav")
require.Error(t, err)
assert.Equal(t, []string{"iguana/kb-whisper"}, *r.calls)
}
func TestEnsemble_ACancelledAutoLanguageJobDoesNotReachTheFallback(t *testing.T) {
r := newEnsemble("base text", "alt text", nil, errors.New("engine down"), true, &verdictAdj{})
ctx, cancel := context.WithCancel(autoCtx())
cancel()
_, err := r.e.Transcribe(ctx, "/a.wav")
require.Error(t, err)
assert.NotContains(t, *r.calls, "berget/kb-whisper", "a stopped job must not ship its audio to a cloud endpoint")
}
+12 -1
View File
@@ -71,9 +71,20 @@ func (g *Gateway) Transcribe(ctx context.Context, path string) (domain.Transcrip
return domain.Transcription{}, fmt.Errorf("transcribe: stat audio: %w", err) return domain.Transcription{}, fmt.Errorf("transcribe: stat audio: %w", err)
} }
// The language is one request field and a closed set: sv is sent, auto sends nothing so
// the engine detects it, and anything else never leaves this process (tapir#61).
fields := [][2]string{{"model", g.model}, {"response_format", "verbose_json"}}
switch lang := domain.LanguageFrom(ctx); lang {
case domain.LangSwedish:
fields = append(fields, [2]string{"language", string(lang)})
case domain.LangAuto:
default:
return domain.Transcription{}, fmt.Errorf("transcribe: language %q is not one tapir sends", lang)
}
var prefix bytes.Buffer var prefix bytes.Buffer
mw := multipart.NewWriter(&prefix) mw := multipart.NewWriter(&prefix)
for _, kv := range [][2]string{{"model", g.model}, {"response_format", "verbose_json"}, {"language", "sv"}} { for _, kv := range fields {
if err := mw.WriteField(kv[0], kv[1]); err != nil { if err := mw.WriteField(kv[0], kv[1]); err != nil {
return domain.Transcription{}, fmt.Errorf("transcribe: build request: %w", err) return domain.Transcription{}, fmt.Errorf("transcribe: build request: %w", err)
} }
@@ -255,3 +255,48 @@ func TestGateway_AnErrorBodyIsBoundedToTwoHundredBytesInTheStatusError(t *testin
require.ErrorAs(t, err, &se) require.ErrorAs(t, err, &se)
assert.LessOrEqual(t, len(se.Snippet), 200, "a gateway's whole body must not become part of an error") assert.LessOrEqual(t, len(se.Snippet), 200, "a gateway's whole body must not become part of an error")
} }
// The language travels on the context (tapir#61); the request field is the only place it
// takes effect, so these observe the request the gateway fake received.
func TestGateway_SendsLanguageSvWhenTheContextCarriesNoneAndForSwedish(t *testing.T) {
for name, ctx := range map[string]context.Context{
"no language in the context": context.Background(),
"swedish": domain.WithLanguage(context.Background(), domain.LangSwedish),
} {
t.Run(name, func(t *testing.T) {
var got seen
srv := serve(t, 200, verboseJSON, &got)
g := transcribe.NewGateway(srv.URL+"/v1", "k", "iguana/kb-whisper", srv.Client())
_, err := g.Transcribe(ctx, wavFile(t, "RIFF-synthetic-pcm"))
require.NoError(t, err)
assert.Equal(t, "sv", got.fields["language"])
})
}
}
func TestGateway_OmitsTheLanguageFieldForAnAutoContextSoTheEngineDetectsIt(t *testing.T) {
var got seen
srv := serve(t, 200, verboseJSON, &got)
g := transcribe.NewGateway(srv.URL+"/v1", "k", "iguana/kb-whisper", srv.Client())
_, err := g.Transcribe(domain.WithLanguage(context.Background(), domain.LangAuto), wavFile(t, "RIFF-synthetic-pcm"))
require.NoError(t, err)
_, present := got.fields["language"]
assert.False(t, present, "an auto upload must not be forced to Swedish")
assert.Equal(t, "iguana/kb-whisper", got.fields["model"], "the rest of the request is unchanged")
}
func TestGateway_ALanguageOutsideTheClosedSetIsRefusedBeforeAnythingIsSent(t *testing.T) {
var got seen
srv := serve(t, 200, verboseJSON, &got)
g := transcribe.NewGateway(srv.URL+"/v1", "k", "iguana/kb-whisper", srv.Client())
_, err := g.Transcribe(domain.WithLanguage(context.Background(), domain.Language("en")), wavFile(t, "RIFF-synthetic-pcm"))
require.Error(t, err)
assert.Empty(t, got.method, "no request may reach the engine with a language the closed set does not hold")
}
@@ -102,3 +102,45 @@ func TestEnsemble_AHungEngineIsASoftFailureAndTheOtherEnginesTextIsUsed(t *testi
assert.Contains(t, got.Reconciliation.Reason, "timeout") assert.Contains(t, got.Reconciliation.Reason, "timeout")
assert.False(t, errors.Is(err, context.DeadlineExceeded)) assert.False(t, errors.Is(err, context.DeadlineExceeded))
} }
// WithTimeout sits in front of both local engines in production (cmd/tapir/uploads.go). It
// must hand the engine the job's context, not a fresh one: the language and the
// cancellation both travel on it (tapir#61).
func TestWithTimeout_TheEngineStillReceivesTheLanguageOfTheJob(t *testing.T) {
for name, tt := range map[string]struct {
ctx context.Context
wantField bool
}{
"auto leaves the field out": {domain.WithLanguage(context.Background(), domain.LangAuto), false},
"swedish sends sv": {domain.WithLanguage(context.Background(), domain.LangSwedish), true},
} {
t.Run(name, func(t *testing.T) {
var got seen
srv := serve(t, 200, verboseJSON, &got)
b := transcribe.WithTimeout(transcribe.NewGateway(srv.URL+"/v1", "k", "iguana/kb-whisper", srv.Client()), time.Minute)
_, err := b.Transcribe(tt.ctx, wavFile(t, "RIFF-synthetic-pcm"))
require.NoError(t, err)
lang, present := got.fields["language"]
assert.Equal(t, tt.wantField, present)
if tt.wantField {
assert.Equal(t, "sv", lang)
}
})
}
}
func TestWithTimeout_ACancelledJobIsNotRestartedByTheWrapper(t *testing.T) {
var got seen
srv := serve(t, 200, verboseJSON, &got)
b := transcribe.WithTimeout(transcribe.NewGateway(srv.URL+"/v1", "k", "iguana/kb-whisper", srv.Client()), time.Minute)
ctx, cancel := context.WithCancel(context.Background())
cancel()
_, err := b.Transcribe(ctx, wavFile(t, "RIFF-synthetic-pcm"))
require.Error(t, err)
assert.Empty(t, got.method, "a cancelled job must not send its audio anywhere through the wrapper")
}
+6 -3
View File
@@ -243,9 +243,12 @@ const (
// never stored in, or merged with, the shared public-caption Transcript (ADR-021). // never stored in, or merged with, the shared public-caption Transcript (ADR-021).
// Content and Segments are empty until Status is UploadDone. // Content and Segments are empty until Status is UploadDone.
type UploadedTranscript struct { type UploadedTranscript struct {
ID string ID string
UserID string UserID string
Title string Title string
// Language is what the upload was (or will be) transcribed as: sv, or auto for
// "Other" (tapir#61).
Language Language
Status UploadStatus Status UploadStatus
Content string Content string
Segments []Segment Segments []Segment
+47
View File
@@ -0,0 +1,47 @@
package domain
import "context"
// Language is the spoken language an upload is transcribed as. It is chosen by the user on
// the upload form and reaches the engines as one request field, so it is a closed set,
// never free text (tapir#61).
type Language string
const (
// LangSwedish sends language=sv: the Swedish engines' own language, and what every
// upload was transcribed as before the choice existed.
LangSwedish Language = "sv"
// LangAuto sends no language, so the engine detects it. It is what "Other" on the form
// means: the Swedish-only engines are not used for it.
LangAuto Language = "auto"
)
// ParseLanguage reads a form value. An empty value is Swedish (a client that predates the
// field); anything that is not exactly a known value is refused rather than normalised.
func ParseLanguage(s string) (Language, bool) {
switch Language(s) {
case "", LangSwedish:
return LangSwedish, true
case LangAuto:
return LangAuto, true
}
return "", false
}
type languageKey struct{}
// WithLanguage returns a context that carries the language of the job it serves. The
// Transcriber interface does not change: engines read it with LanguageFrom, the same way
// they read the progress Reporter (ADR-035).
func WithLanguage(ctx context.Context, l Language) context.Context {
return context.WithValue(ctx, languageKey{}, l)
}
// LanguageFrom returns the language carried by ctx, or Swedish when it carries none, so a
// caller that sets nothing behaves as it always did.
func LanguageFrom(ctx context.Context) Language {
if l, ok := ctx.Value(languageKey{}).(Language); ok && l != "" {
return l
}
return LangSwedish
}
+41
View File
@@ -0,0 +1,41 @@
package domain_test
import (
"context"
"testing"
"github.com/stretchr/testify/assert"
"git.d-ma.be/mathias/tapir/internal/domain"
)
func TestParseLanguage_KnownValuesAndEmptyAreAcceptedEverythingElseIsNot(t *testing.T) {
tests := []struct {
in string
want domain.Language
wantOK bool
}{
{"sv", domain.LangSwedish, true},
{"auto", domain.LangAuto, true},
{"", domain.LangSwedish, true}, // a client that sends no field keeps today's behaviour
{"SV", "", false}, // exact match: the value reaches a request field
{" sv", "", false},
{"sv\n", "", false},
{"en", "", false}, // a specific other language is not offered yet
{"sv,en", "", false},
{"../sv", "", false},
}
for _, tt := range tests {
t.Run(tt.in, func(t *testing.T) {
got, ok := domain.ParseLanguage(tt.in)
assert.Equal(t, tt.wantOK, ok)
assert.Equal(t, tt.want, got)
})
}
}
func TestLanguageFrom_IsSwedishWhenTheContextCarriesNoneAndOtherwiseWhatWasSet(t *testing.T) {
assert.Equal(t, domain.LangSwedish, domain.LanguageFrom(context.Background()))
ctx := domain.WithLanguage(context.Background(), domain.LangAuto)
assert.Equal(t, domain.LangAuto, domain.LanguageFrom(ctx))
}
+7 -4
View File
@@ -25,7 +25,7 @@ type Prober interface {
// Store is the slice of the private transcript store the intake needs. // Store is the slice of the private transcript store the intake needs.
type Store interface { type Store interface {
CreateUpload(ctx context.Context, userID, title string, hadVideo bool) (string, error) CreateUpload(ctx context.Context, userID, title string, lang domain.Language, hadVideo bool) (string, error)
FailUpload(ctx context.Context, userID, id string, kind domain.UploadErrorKind) error FailUpload(ctx context.Context, userID, id string, kind domain.UploadErrorKind) error
} }
@@ -35,6 +35,9 @@ type Job struct {
UserID string UserID string
UploadID string UploadID string
Path string Path string
// Language is what the user chose for this recording; empty means Swedish (a job
// queued before the choice existed).
Language domain.Language
// Duration is what the file CLAIMED when probed. The worker compares it with what // Duration is what the file CLAIMED when probed. The worker compares it with what
// was really converted; zero means unknown (ADR-035 review). // was really converted; zero means unknown (ADR-035 review).
Duration time.Duration Duration time.Duration
@@ -135,7 +138,7 @@ const busyMessage = "Too many recordings are waiting to be transcribed. Please t
// validation yields a *RejectError and creates NO row — rejection is synchronous // validation yields a *RejectError and creates NO row — rejection is synchronous
// and leaves nothing behind. On success the temp file is handed to the job and the // and leaves nothing behind. On success the temp file is handed to the job and the
// worker deletes it; on every failure path Accept deletes it itself. // worker deletes it; on every failure path Accept deletes it itself.
func (in *Intake) Accept(ctx context.Context, userID, title string, src io.Reader) (string, error) { func (in *Intake) Accept(ctx context.Context, userID, title string, lang domain.Language, src io.Reader) (string, error) {
// Refuse before spooling when the queue has no room, if the submitter can say so: a full // Refuse before spooling when the queue has no room, if the submitter can say so: a full
// queue otherwise cost up to the upload cap on disk, an ffprobe run and a failed row for // queue otherwise cost up to the upload cap on disk, an ffprobe run and a failed row for
// every refused attempt. // every refused attempt.
@@ -190,11 +193,11 @@ func (in *Intake) Accept(ctx context.Context, userID, title string, src io.Reade
return "", err return "", err
} }
id, err := in.store.CreateUpload(ctx, userID, title, probe.HasVideo) id, err := in.store.CreateUpload(ctx, userID, title, lang, probe.HasVideo)
if err != nil { if err != nil {
return "", fmt.Errorf("uploads: create: %w", err) return "", fmt.Errorf("uploads: create: %w", err)
} }
if err := in.sub.Submit(Job{UserID: userID, UploadID: id, Path: path, Duration: probe.Duration}); err != nil { if err := in.sub.Submit(Job{UserID: userID, UploadID: id, Path: path, Language: lang, Duration: probe.Duration}); err != nil {
if errors.Is(err, ErrQueueFull) { if errors.Is(err, ErrQueueFull) {
// The user's to retry, not a server fault: refuse it as such, and do not leave // The user's to retry, not a server fault: refuse it as such, and do not leave
// the row queued -- that is the stuck spinner. // the row queued -- that is the stuck spinner.
+5 -5
View File
@@ -23,7 +23,7 @@ func TestAccept_AFullQueueIsABusyRejectionNotAServerErrorAndLeavesNothingBehind(
r := newRig(t) r := newRig(t)
r.sub.err = uploads.ErrQueueFull r.sub.err = uploads.ErrQueueFull
_, err := r.in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data"))
var rej *uploads.RejectError var rej *uploads.RejectError
require.True(t, errors.As(err, &rej), "a full queue must be a rejection the web layer can answer with 429: %v", err) require.True(t, errors.As(err, &rej), "a full queue must be a rejection the web layer can answer with 429: %v", err)
@@ -48,7 +48,7 @@ func TestAccept_ANoRoomQueueIsRefusedBeforeAnythingIsSpooledProbedOrStored(t *te
full := &roomSubmitter{room: false} full := &roomSubmitter{room: false}
in := uploads.NewIntake(r.st, r.pr, full, uploads.Limits{MaxBytes: 1024, MaxDuration: 90 * time.Minute}, r.tmp) in := uploads.NewIntake(r.st, r.pr, full, uploads.Limits{MaxBytes: 1024, MaxDuration: 90 * time.Minute}, r.tmp)
_, err := in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) _, err := in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data"))
var rej *uploads.RejectError var rej *uploads.RejectError
require.True(t, errors.As(err, &rej)) require.True(t, errors.As(err, &rej))
@@ -62,7 +62,7 @@ func TestAccept_ANoRoomQueueIsRefusedBeforeAnythingIsSpooledProbedOrStored(t *te
func TestAccept_ASubmitterThatCannotSayStillWorksAsBefore(t *testing.T) { func TestAccept_ASubmitterThatCannotSayStillWorksAsBefore(t *testing.T) {
r := newRig(t) r := newRig(t)
_, err := r.in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data"))
require.NoError(t, err, "a Submitter without HasRoom is accepted: the check at Submit remains") require.NoError(t, err, "a Submitter without HasRoom is accepted: the check at Submit remains")
} }
@@ -81,7 +81,7 @@ func (b *brokenBody) Read(p []byte) (int, error) {
func TestAccept_AnErrorReadingTheClientsBodyIsMarkedAsTheClientsAndAWrappedSubmitErrorIsNot(t *testing.T) { func TestAccept_AnErrorReadingTheClientsBodyIsMarkedAsTheClientsAndAWrappedSubmitErrorIsNot(t *testing.T) {
r := newRig(t) r := newRig(t)
_, err := r.in.Accept(context.Background(), "user-1", "t", &brokenBody{n: 1}) _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, &brokenBody{n: 1})
require.Error(t, err) require.Error(t, err)
assert.ErrorIs(t, err, uploads.ErrClientRead, "a dropped client body is the sender's to retry") assert.ErrorIs(t, err, uploads.ErrClientRead, "a dropped client body is the sender's to retry")
@@ -90,7 +90,7 @@ func TestAccept_AnErrorReadingTheClientsBodyIsMarkedAsTheClientsAndAWrappedSubmi
// A fault that is ours must NOT carry the marker, even if it wraps the same sentinel. // A fault that is ours must NOT carry the marker, even if it wraps the same sentinel.
r2 := newRig(t) r2 := newRig(t)
r2.sub.err = fmt.Errorf("db: %w", io.ErrUnexpectedEOF) r2.sub.err = fmt.Errorf("db: %w", io.ErrUnexpectedEOF)
_, err = r2.in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) _, err = r2.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data"))
require.Error(t, err) require.Error(t, err)
assert.NotErrorIs(t, err, uploads.ErrClientRead) assert.NotErrorIs(t, err, uploads.ErrClientRead)
} }
+7 -6
View File
@@ -15,6 +15,7 @@ import (
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"git.d-ma.be/mathias/tapir/internal/domain"
"git.d-ma.be/mathias/tapir/internal/uploads" "git.d-ma.be/mathias/tapir/internal/uploads"
) )
@@ -31,7 +32,7 @@ func startSlowReceive(t *testing.T, r *rig, in *uploads.Intake, user string, wan
pr, pw := io.Pipe() pr, pw := io.Pipe()
s := &slowReceive{w: pw, done: make(chan error, 1)} s := &slowReceive{w: pw, done: make(chan error, 1)}
go func() { go func() {
_, err := in.Accept(context.Background(), user, "t", pr) _, err := in.Accept(context.Background(), user, "t", domain.LangSwedish, pr)
s.done <- err s.done <- err
}() }()
require.Eventually(t, func() bool { return len(r.tmpFiles(t)) == wantFiles }, 2*time.Second, 5*time.Millisecond, require.Eventually(t, func() bool { return len(r.tmpFiles(t)) == wantFiles }, 2*time.Second, 5*time.Millisecond,
@@ -80,7 +81,7 @@ func TestAccept_APerUserLimitOnConcurrentReceivesRefusesTheNextBodyBeforeSpoolin
startSlowReceive(t, r, in, "user-1", 1) startSlowReceive(t, r, in, "user-1", 1)
startSlowReceive(t, r, in, "user-1", 2) startSlowReceive(t, r, in, "user-1", 2)
_, err := in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) _, err := in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data"))
requireBusy(t, err) requireBusy(t, err)
assert.Len(t, r.tmpFiles(t), 2, "the refused body was never spooled") assert.Len(t, r.tmpFiles(t), 2, "the refused body was never spooled")
@@ -98,7 +99,7 @@ func TestAccept_AnOverallLimitOnConcurrentReceivesRefusesTheNextUser(t *testing.
startSlowReceive(t, r, in, u, i+1) startSlowReceive(t, r, in, u, i+1)
} }
_, err := in.Accept(context.Background(), "d", "t", strings.NewReader("data")) _, err := in.Accept(context.Background(), "d", "t", domain.LangSwedish, strings.NewReader("data"))
requireBusy(t, err) requireBusy(t, err)
assert.Len(t, r.tmpFiles(t), 3) assert.Len(t, r.tmpFiles(t), 3)
@@ -112,7 +113,7 @@ func TestAccept_AFinishedReceiveFreesItsSlot(t *testing.T) {
first := startSlowReceive(t, r, in, "user-1", 1) first := startSlowReceive(t, r, in, "user-1", 1)
require.NoError(t, first.finish(t)) require.NoError(t, first.finish(t))
_, err := in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) _, err := in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data"))
require.NoError(t, err, "the slot is released when Accept returns") require.NoError(t, err, "the slot is released when Accept returns")
} }
@@ -122,10 +123,10 @@ func TestAccept_AFailedReceiveFreesItsSlot(t *testing.T) {
in := uploads.NewIntake(r.st, r.pr, r.sub, uploads.Limits{MaxBytes: 1024, MaxDuration: time.Hour}, r.tmp) in := uploads.NewIntake(r.st, r.pr, r.sub, uploads.Limits{MaxBytes: 1024, MaxDuration: time.Hour}, r.tmp)
in.MaxReceivesPerUser = 1 in.MaxReceivesPerUser = 1
in.MaxReceives = 5 in.MaxReceives = 5
_, err := in.Accept(context.Background(), "user-1", "t", &brokenBody{n: 1}) _, err := in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, &brokenBody{n: 1})
require.ErrorIs(t, err, uploads.ErrClientRead) require.ErrorIs(t, err, uploads.ErrClientRead)
_, err = in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) _, err = in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data"))
require.NoError(t, err, "a dropped client must not leak a slot") require.NoError(t, err, "a dropped client must not leak a slot")
} }
+3 -2
View File
@@ -12,6 +12,7 @@ import (
"testing" "testing"
"time" "time"
"git.d-ma.be/mathias/tapir/internal/domain"
"git.d-ma.be/mathias/tapir/internal/uploads" "git.d-ma.be/mathias/tapir/internal/uploads"
) )
@@ -21,11 +22,11 @@ func TestNewIntake_DefaultsAllowTwoReceivesPerUserAndFourOverall(t *testing.T) {
startSlowReceive(t, r, in, "a", 1) startSlowReceive(t, r, in, "a", 1)
startSlowReceive(t, r, in, "a", 2) startSlowReceive(t, r, in, "a", 2)
_, err := in.Accept(context.Background(), "a", "t", strings.NewReader("x")) _, err := in.Accept(context.Background(), "a", "t", domain.LangSwedish, strings.NewReader("x"))
requireBusy(t, err) // a third body from one user requireBusy(t, err) // a third body from one user
startSlowReceive(t, r, in, "b", 3) startSlowReceive(t, r, in, "b", 3)
startSlowReceive(t, r, in, "b", 4) startSlowReceive(t, r, in, "b", 4)
_, err = in.Accept(context.Background(), "c", "t", strings.NewReader("x")) _, err = in.Accept(context.Background(), "c", "t", domain.LangSwedish, strings.NewReader("x"))
requireBusy(t, err) // a fifth body overall requireBusy(t, err) // a fifth body overall
} }
+24 -11
View File
@@ -21,13 +21,15 @@ import (
type fakeStore struct { type fakeStore struct {
created []string // titles created []string // titles
langs []domain.Language
hadVideo []bool hadVideo []bool
failed map[string]domain.UploadErrorKind failed map[string]domain.UploadErrorKind
nextID string nextID string
} }
func (f *fakeStore) CreateUpload(_ context.Context, _, title string, hadVideo bool) (string, error) { func (f *fakeStore) CreateUpload(_ context.Context, _, title string, lang domain.Language, hadVideo bool) (string, error) {
f.created = append(f.created, title) f.created = append(f.created, title)
f.langs = append(f.langs, lang)
f.hadVideo = append(f.hadVideo, hadVideo) f.hadVideo = append(f.hadVideo, hadVideo)
return f.nextID, nil return f.nextID, nil
} }
@@ -98,7 +100,7 @@ func TestAccept_ValidAudioIsStoredProbedQueuedAndSubmitted(t *testing.T) {
r := newRig(t) r := newRig(t)
audio := bytes.Repeat([]byte("a"), 500) audio := bytes.Repeat([]byte("a"), 500)
id, err := r.in.Accept(context.Background(), "user-1", "Teams export", bytes.NewReader(audio)) id, err := r.in.Accept(context.Background(), "user-1", "Teams export", domain.LangSwedish, bytes.NewReader(audio))
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, "up-1", id) assert.Equal(t, "up-1", id)
@@ -106,7 +108,7 @@ func TestAccept_ValidAudioIsStoredProbedQueuedAndSubmitted(t *testing.T) {
assert.Equal(t, audio, r.pr.seen, "the prober must see the complete uploaded bytes, on disk") assert.Equal(t, audio, r.pr.seen, "the prober must see the complete uploaded bytes, on disk")
require.Len(t, r.sub.jobs, 1) require.Len(t, r.sub.jobs, 1)
job := r.sub.jobs[0] job := r.sub.jobs[0]
assert.Equal(t, uploads.Job{UserID: "user-1", UploadID: "up-1", Path: job.Path, Duration: 10 * time.Minute}, job, assert.Equal(t, uploads.Job{UserID: "user-1", UploadID: "up-1", Path: job.Path, Duration: 10 * time.Minute, Language: domain.LangSwedish}, job,
"the job carries what the file CLAIMED, so the worker can compare it with what was converted") "the job carries what the file CLAIMED, so the worker can compare it with what was converted")
assert.Equal(t, filepath.Dir(job.Path), r.tmp, "the temp file lives in the configured directory") assert.Equal(t, filepath.Dir(job.Path), r.tmp, "the temp file lives in the configured directory")
got, err := os.ReadFile(job.Path) got, err := os.ReadFile(job.Path)
@@ -117,7 +119,7 @@ func TestAccept_ValidAudioIsStoredProbedQueuedAndSubmitted(t *testing.T) {
func TestAccept_OversizeBodyIsRejectedAndLeavesNothingBehind(t *testing.T) { func TestAccept_OversizeBodyIsRejectedAndLeavesNothingBehind(t *testing.T) {
r := newRig(t) r := newRig(t)
_, err := r.in.Accept(context.Background(), "user-1", "big", strings.NewReader(strings.Repeat("a", 1025))) _, err := r.in.Accept(context.Background(), "user-1", "big", domain.LangSwedish, strings.NewReader(strings.Repeat("a", 1025)))
var rej *uploads.RejectError var rej *uploads.RejectError
require.True(t, errors.As(err, &rej)) require.True(t, errors.As(err, &rej))
@@ -132,7 +134,7 @@ func TestAccept_OversizeBodyIsRejectedAndLeavesNothingBehind(t *testing.T) {
func TestAccept_BodyExactlyAtTheLimitIsAccepted(t *testing.T) { func TestAccept_BodyExactlyAtTheLimitIsAccepted(t *testing.T) {
r := newRig(t) r := newRig(t)
_, err := r.in.Accept(context.Background(), "user-1", "edge", strings.NewReader(strings.Repeat("a", 1024))) _, err := r.in.Accept(context.Background(), "user-1", "edge", domain.LangSwedish, strings.NewReader(strings.Repeat("a", 1024)))
require.NoError(t, err) require.NoError(t, err)
} }
@@ -152,7 +154,7 @@ func TestAccept_ProbeVerdictRejectsWithoutRowAndRemovesTheFile(t *testing.T) {
r := newRig(t) r := newRig(t)
r.pr.probe = tt.probe r.pr.probe = tt.probe
_, err := r.in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data"))
var rej *uploads.RejectError var rej *uploads.RejectError
require.True(t, errors.As(err, &rej)) require.True(t, errors.As(err, &rej))
@@ -169,7 +171,7 @@ func TestAccept_UnreadableFileIsARejectionButAProberFaultIsAnError(t *testing.T)
r := newRig(t) r := newRig(t)
r.pr.err = uploads.ErrUnreadable r.pr.err = uploads.ErrUnreadable
_, err := r.in.Accept(context.Background(), "user-1", "t", strings.NewReader("not media")) _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("not media"))
var rej *uploads.RejectError var rej *uploads.RejectError
require.True(t, errors.As(err, &rej), "the user's file is the problem: tell them") require.True(t, errors.As(err, &rej), "the user's file is the problem: tell them")
@@ -180,7 +182,7 @@ func TestAccept_UnreadableFileIsARejectionButAProberFaultIsAnError(t *testing.T)
r := newRig(t) r := newRig(t)
r.pr.err = errors.New("exec: ffprobe: executable file not found") r.pr.err = errors.New("exec: ffprobe: executable file not found")
_, err := r.in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data"))
require.Error(t, err) require.Error(t, err)
var rej *uploads.RejectError var rej *uploads.RejectError
@@ -194,7 +196,7 @@ func TestAccept_ClientDisconnectMidUploadLeavesNothingBehind(t *testing.T) {
r := newRig(t) r := newRig(t)
body := io.MultiReader(strings.NewReader("partial"), iotest.ErrReader(errors.New("client went away"))) body := io.MultiReader(strings.NewReader("partial"), iotest.ErrReader(errors.New("client went away")))
_, err := r.in.Accept(context.Background(), "user-1", "t", body) _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, body)
require.Error(t, err) require.Error(t, err)
assert.Empty(t, r.pr.paths) assert.Empty(t, r.pr.paths)
@@ -206,7 +208,7 @@ func TestAccept_SubmitFailureFailsTheRowHonestlyAndRemovesTheFile(t *testing.T)
r := newRig(t) r := newRig(t)
r.sub.err = errors.New("worker stopped") r.sub.err = errors.New("worker stopped")
_, err := r.in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data"))
require.Error(t, err) require.Error(t, err)
assert.Equal(t, domain.UploadErrInterrupted, r.st.failed["up-1"], assert.Equal(t, domain.UploadErrInterrupted, r.st.failed["up-1"],
@@ -228,7 +230,7 @@ func TestAccept_AFileWithAVideoTrackIsAcceptedAndRecordsThatItHadOne(t *testing.
r := newRig(t) r := newRig(t)
r.pr.probe = tt.probe r.pr.probe = tt.probe
id, err := r.in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) id, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data"))
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, "up-1", id) assert.Equal(t, "up-1", id)
@@ -238,3 +240,14 @@ func TestAccept_AFileWithAVideoTrackIsAcceptedAndRecordsThatItHadOne(t *testing.
}) })
} }
} }
func TestAccept_TheChosenLanguageIsStoredAndHandedToTheJob(t *testing.T) {
r := newRig(t)
_, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangAuto, strings.NewReader("data"))
require.NoError(t, err)
assert.Equal(t, []domain.Language{domain.LangAuto}, r.st.langs, "the row records what the user chose")
require.Len(t, r.sub.jobs, 1)
assert.Equal(t, domain.LangAuto, r.sub.jobs[0].Language, "the worker is told, so the engines are")
}
+15 -2
View File
@@ -33,6 +33,13 @@ type Transcriber interface {
Transcribe(ctx context.Context, wavPath string) (domain.Transcription, error) Transcribe(ctx context.Context, wavPath string) (domain.Transcription, error)
} }
// LanguagePlanner is a Planner whose phases depend on the language of the job: a recording
// that is not Swedish skips the second engine and the reconcile, and the page should not list
// them as coming (tapir#61). The worker prefers it over Planner.
type LanguagePlanner interface {
PlanFor(lang domain.Language) []progress.Planned
}
// Planner is an optional extra on a Transcriber: it declares its own phases (for // Planner is an optional extra on a Transcriber: it declares its own phases (for
// example one per engine) so the page can show them. A Transcriber that does not // example one per engine) so the page can show them. A Transcriber that does not
// implement it gets a single "transcribing" phase from the worker. Whatever a Planner // implement it gets a single "transcribing" phase from the worker. Whatever a Planner
@@ -251,8 +258,14 @@ func (w *Worker) Process(ctx context.Context, job Job) {
} }
plan := []progress.Planned{{Key: "received", Label: "Received"}, {Key: "audio", Label: "Preparing the audio"}} plan := []progress.Planned{{Key: "received", Label: "Received"}, {Key: "audio", Label: "Preparing the audio"}}
lang := job.Language
if lang == "" {
lang = domain.LangSwedish // queued before the choice existed
}
planner, planned := w.tr.(Planner) planner, planned := w.tr.(Planner)
if planned { if lp, ok := w.tr.(LanguagePlanner); ok {
plan = append(plan, lp.PlanFor(lang)...)
} else if planned {
plan = append(plan, planner.Plan()...) plan = append(plan, planner.Plan()...)
} else { } else {
plan = append(plan, progress.Planned{Key: "transcribe", Label: "Transcribing"}) plan = append(plan, progress.Planned{Key: "transcribe", Label: "Transcribing"})
@@ -313,7 +326,7 @@ func (w *Worker) Process(ctx context.Context, job Job) {
if !planned { if !planned {
tr.Begin("transcribe", 0) tr.Begin("transcribe", 0)
} }
t, err := w.tr.Transcribe(progress.With(jobCtx, tr), wav) t, err := w.tr.Transcribe(domain.WithLanguage(progress.With(jobCtx, tr), lang), wav)
if err != nil { if err != nil {
log.Error("upload: transcribe", "kind", errkind.Of(err), "elapsed", time.Since(start).Round(time.Second)) log.Error("upload: transcribe", "kind", errkind.Of(err), "elapsed", time.Since(start).Round(time.Second))
abort(domain.UploadErrTranscriptionFailed) abort(domain.UploadErrTranscriptionFailed)
+44
View File
@@ -213,3 +213,47 @@ func TestWorkerProgress_AStoreFailureLogsItsSQLStateNeverTheValueItQuotes(t *tes
assert.Contains(t, out, "upload: store transcript") assert.Contains(t, out, "upload: store transcript")
assert.Contains(t, out, "kind=sql_22P02") assert.Contains(t, out, "kind=sql_22P02")
} }
// langPlannedTr plans per language, like the Ensemble: an auto job has no base engine.
type langPlannedTr struct{ *fakeTranscriber }
func (langPlannedTr) Plan() []progress.Planned { return plannedTr{}.Plan() }
func (langPlannedTr) PlanFor(l domain.Language) []progress.Planned {
if l == domain.LangAuto {
return []progress.Planned{{Key: "engine_b", Label: "Transcribing with B"}}
}
return plannedTr{}.Plan()
}
func (p langPlannedTr) Transcribe(ctx context.Context, wav string) (domain.Transcription, error) {
r := progress.From(ctx)
if domain.LanguageFrom(ctx) != domain.LangAuto {
r.Begin("engine_a", 0)
r.Done("engine_a")
}
r.Begin("engine_b", 0)
r.Done("engine_b")
return p.fakeTranscriber.Transcribe(ctx, wav)
}
func TestWorkerProgress_AJobGetsThePlanOfItsLanguage(t *testing.T) {
for name, tt := range map[string]struct {
lang domain.Language
want []string
}{
"auto": {domain.LangAuto, []string{"received:done", "audio:done", "engine_b:done", "save:done"}},
"swedish": {domain.LangSwedish, []string{"received:done", "audio:done", "engine_a:done", "engine_b:done", "save:done"}},
"no language (queued before the choice existed)": {"", []string{"received:done", "audio:done", "engine_a:done", "engine_b:done", "save:done"}},
} {
t.Run(name, func(t *testing.T) {
r := newWRig(t, time.Minute)
w := newProgressWorker(t, r, langPlannedTr{r.tr})
job := r.spool(t, "a")
job.Language = tt.lang
w.Process(context.Background(), job)
assert.Equal(t, tt.want, keysAndStates(r.st.lastSteps()))
})
}
}
+23
View File
@@ -126,6 +126,7 @@ type fakeTranscriber struct {
result domain.Transcription result domain.Transcription
err error err error
paths []string paths []string
langs []domain.Language // what each call's context carried
// block, when non-nil, holds every call until it is closed or ctx ends. // block, when non-nil, holds every call until it is closed or ctx ends.
block chan struct{} block chan struct{}
running atomic.Int32 running atomic.Int32
@@ -134,6 +135,7 @@ type fakeTranscriber struct {
func (f *fakeTranscriber) Transcribe(ctx context.Context, wav string) (domain.Transcription, error) { func (f *fakeTranscriber) Transcribe(ctx context.Context, wav string) (domain.Transcription, error) {
f.paths = append(f.paths, wav) f.paths = append(f.paths, wav)
f.langs = append(f.langs, domain.LanguageFrom(ctx))
n := f.running.Add(1) n := f.running.Add(1)
defer f.running.Add(-1) defer f.running.Add(-1)
for { for {
@@ -337,3 +339,24 @@ func (dbErr) Error() string {
return `ERROR: invalid input syntax for type json: "zebrakaka secret words"` return `ERROR: invalid input syntax for type json: "zebrakaka secret words"`
} }
func (dbErr) SQLState() string { return "22P02" } func (dbErr) SQLState() string { return "22P02" }
func TestWorker_TheTranscriberSeesTheLanguageOfTheJobOnItsContext(t *testing.T) {
for name, tt := range map[string]struct {
job domain.Language
want domain.Language
}{
"auto": {domain.LangAuto, domain.LangAuto},
"swedish": {domain.LangSwedish, domain.LangSwedish},
"a job with no language": {"", domain.LangSwedish}, // queued before the field existed
} {
t.Run(name, func(t *testing.T) {
r := newWRig(t, time.Minute)
job := r.spool(t, "a")
job.Language = tt.job
r.w.Process(context.Background(), job)
assert.Equal(t, []domain.Language{tt.want}, r.tr.langs)
})
}
}
+31 -2
View File
@@ -28,7 +28,7 @@ import (
// *uploads.Intake satisfies it. It is the whole of what the handlers know about // *uploads.Intake satisfies it. It is the whole of what the handlers know about
// validation, spooling and queueing. // validation, spooling and queueing.
type Uploader interface { type Uploader interface {
Accept(ctx context.Context, userID, title string, src io.Reader) (id string, err error) Accept(ctx context.Context, userID, title string, lang domain.Language, src io.Reader) (id string, err error)
} }
const ( const (
@@ -36,6 +36,8 @@ const (
// and the title field; the cap itself is enforced on the file part by Intake. // and the title field; the cap itself is enforced on the file part by Intake.
uploadOverhead = 1 << 20 uploadOverhead = 1 << 20
maxTitleRunes = 200 maxTitleRunes = 200
// maxLanguageBytes bounds what is read of the language field; a valid value is at most four bytes, so a longer one fails the parse.
maxLanguageBytes = 16
// uploadPollEvery is how often an unfinished upload re-asks for its status. // uploadPollEvery is how often an unfinished upload re-asks for its status.
uploadPollEvery = "2s" uploadPollEvery = "2s"
// serviceSilentAfter is how long a job may go without any signal (heartbeat or // serviceSilentAfter is how long a job may go without any signal (heartbeat or
@@ -89,6 +91,7 @@ func (a *App) handleUploadPost(w http.ResponseWriter, r *http.Request) {
return return
} }
title := "" title := ""
lang := domain.LangSwedish
for { for {
part, err := mr.NextPart() part, err := mr.NextPart()
if errors.Is(err, io.EOF) { if errors.Is(err, io.EOF) {
@@ -121,11 +124,29 @@ func (a *App) handleUploadPost(w http.ResponseWriter, r *http.Request) {
return return
} }
title = cleanTitle(string(b)) title = cleanTitle(string(b))
case "language":
// A closed set read before the file: anything else is refused before the file
// is spooled, so a crafted value never reaches an engine (tapir#61).
b, err := io.ReadAll(io.LimitReader(part, maxLanguageBytes+1))
if err != nil {
if a.abandoned(w, r) {
return
}
a.renderUploadsPage(w, r, userID, http.StatusBadRequest, "That upload could not be read. Please try again.")
return
}
l, ok := domain.ParseLanguage(string(b))
if !ok {
a.logger().Info("upload rejected", "reason", "language")
a.renderUploadsPage(w, r, userID, http.StatusBadRequest, "Choose a language from the list.")
return
}
lang = l
case "file": case "file":
if title == "" { if title == "" {
title = titleFromFilename(part.FileName()) title = titleFromFilename(part.FileName())
} }
id, err := a.Uploads.Accept(r.Context(), userID, title, part) id, err := a.Uploads.Accept(r.Context(), userID, title, lang, part)
if err != nil { if err != nil {
a.uploadFailure(w, r, userID, err) a.uploadFailure(w, r, userID, err)
return return
@@ -300,6 +321,14 @@ func cloudProvider(endpoint string) (string, bool) {
return "", false return "", false
} }
// languageLabel is how the page names the language a recording was transcribed as.
func languageLabel(l domain.Language) string {
if l == domain.LangAuto {
return "Other (detected by the engine)"
}
return "Swedish"
}
// 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 {
+9
View File
@@ -24,6 +24,14 @@ templ UploadsPage(v uploadsView) {
} }
<form method="post" action="/uploads" enctype="multipart/form-data"> <form method="post" action="/uploads" enctype="multipart/form-data">
<p><label>Title (optional) <input type="text" name="title" maxlength="200"/></label></p> <p><label>Title (optional) <input type="text" name="title" maxlength="200"/></label></p>
<p>
<label>Language
<select name="language">
<option value="sv" selected>Swedish</option>
<option value="auto">Other (detected by the engine)</option>
</select>
</label>
</p>
<p><input type="file" name="file" accept="audio/*,.m4a,.mp3,.wav,.opus,.ogg,.mp4" required/></p> <p><input type="file" name="file" accept="audio/*,.m4a,.mp3,.wav,.opus,.ogg,.mp4" required/></p>
<p><button type="submit">Upload and transcribe</button></p> <p><button type="submit">Upload and transcribe</button></p>
</form> </form>
@@ -49,6 +57,7 @@ templ UploadPage(u domain.UploadedTranscript, chatEnabled bool, now time.Time) {
@Layout("Tapir — " + u.Title) { @Layout("Tapir — " + u.Title) {
<article class="account"> <article class="account">
<h1>{ u.Title }</h1> <h1>{ u.Title }</h1>
<p class="muted">Language: { languageLabel(u.Language) }</p>
@uploadStatus(u, chatEnabled, now) @uploadStatus(u, chatEnabled, now)
<p><a href="/uploads">All recordings</a></p> <p><a href="/uploads">All recordings</a></p>
</article> </article>
@@ -14,6 +14,8 @@ import (
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"git.d-ma.be/mathias/tapir/internal/domain"
) )
// A client that hangs up in the middle of an upload is routine, not a server fault. The // A client that hangs up in the middle of an upload is routine, not a server fault. The
@@ -85,7 +87,7 @@ func TestUploadPost_AClientThatHangsUpAnywhereInTheBodyIsLoggedAtInfoAndNeverAsA
// context is cancelled, then Accept fails the way a read from a dead connection does. // context is cancelled, then Accept fails the way a read from a dead connection does.
type goneMidAccept struct{ cancel context.CancelFunc } type goneMidAccept struct{ cancel context.CancelFunc }
func (g goneMidAccept) Accept(context.Context, string, string, io.Reader) (string, error) { func (g goneMidAccept) Accept(context.Context, string, string, domain.Language, io.Reader) (string, error) {
g.cancel() g.cancel()
return "", fmt.Errorf("receive: %w", io.ErrClosedPipe) return "", fmt.Errorf("receive: %w", io.ErrClosedPipe)
} }
+2 -1
View File
@@ -16,6 +16,7 @@ import (
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"git.d-ma.be/mathias/tapir/internal/domain"
"git.d-ma.be/mathias/tapir/internal/uploads" "git.d-ma.be/mathias/tapir/internal/uploads"
"git.d-ma.be/mathias/tapir/internal/web" "git.d-ma.be/mathias/tapir/internal/web"
) )
@@ -132,7 +133,7 @@ func TestUploadPost_AcceptedAudioQueuesARowAndRedirectsToIt(t *testing.T) {
assert.Equal(t, "Sprint review", title) assert.Equal(t, "Sprint review", title)
assert.Equal(t, "/uploads/"+id, rec.Header().Get("Location")) assert.Equal(t, "/uploads/"+id, rec.Header().Get("Location"))
require.Len(t, sub.jobs, 1) require.Len(t, sub.jobs, 1)
assert.Equal(t, uploads.Job{UserID: userID, UploadID: id, Path: sub.jobs[0].Path, Duration: audioOK.Duration}, sub.jobs[0]) assert.Equal(t, uploads.Job{UserID: userID, UploadID: id, Path: sub.jobs[0].Path, Language: domain.LangSwedish, Duration: audioOK.Duration}, sub.jobs[0])
spooled, err := os.ReadFile(sub.jobs[0].Path) spooled, err := os.ReadFile(sub.jobs[0].Path)
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, "synthetic audio bytes", string(spooled)) assert.Equal(t, "synthetic audio bytes", string(spooled))
@@ -0,0 +1,151 @@
package web_test
// The language of a recording is chosen on the upload form (tapir#61). These drive the real
// Router, Intake and Postgres store; the submitter is the process boundary and records the
// job the worker would receive.
import (
"bytes"
"context"
"mime/multipart"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"git.d-ma.be/mathias/tapir/internal/domain"
)
// langReq builds an upload whose parts come in the order given: "title", "language" and
// "file" (the file part always carries the same synthetic bytes).
func langReq(t *testing.T, order []string, language string) *http.Request {
t.Helper()
var buf bytes.Buffer
mw := multipart.NewWriter(&buf)
for _, part := range order {
switch part {
case "title":
require.NoError(t, mw.WriteField("title", "m"))
case "language":
require.NoError(t, mw.WriteField("language", language))
case "file":
fw, err := mw.CreateFormFile("file", "m.wav")
require.NoError(t, err)
_, err = fw.Write([]byte("RIFF....WAVE"))
require.NoError(t, err)
}
}
require.NoError(t, mw.Close())
req := httptest.NewRequest(http.MethodPost, "/uploads", &buf)
req.Header.Set("Content-Type", mw.FormDataContentType())
return req
}
func rowLanguage(t *testing.T, location string) domain.Language {
t.Helper()
id := strings.TrimPrefix(location, "/uploads/")
got, err := newStore(t).GetUpload(context.Background(), userID, id)
require.NoError(t, err)
return got.Language
}
func TestUploadPost_TheLanguageFieldReachesTheJobAndTheRow(t *testing.T) {
for _, tt := range []struct {
name string
language string
want domain.Language
}{
{"other", "auto", domain.LangAuto},
{"swedish", "sv", domain.LangSwedish},
} {
t.Run(tt.name, func(t *testing.T) {
app, sub := newUploadApp(t, audioOK, 200<<20)
rec := do(t, app, langReq(t, []string{"title", "language", "file"}, tt.language))
require.Equal(t, http.StatusSeeOther, rec.Code)
require.Len(t, sub.jobs, 1)
assert.Equal(t, tt.want, sub.jobs[0].Language)
assert.Equal(t, tt.want, rowLanguage(t, rec.Header().Get("Location")))
})
}
}
func TestUploadPost_NoLanguageFieldIsSwedishAsBefore(t *testing.T) {
app, sub := newUploadApp(t, audioOK, 200<<20)
rec := do(t, app, langReq(t, []string{"title", "file"}, ""))
require.Equal(t, http.StatusSeeOther, rec.Code)
require.Len(t, sub.jobs, 1)
assert.Equal(t, domain.LangSwedish, sub.jobs[0].Language)
assert.Equal(t, domain.LangSwedish, rowLanguage(t, rec.Header().Get("Location")))
}
// A part after the file is never read (the file part ends the stream), so a client that
// orders its fields wrongly gets Swedish, the old behaviour, not a guess.
func TestUploadPost_ALanguageFieldAfterTheFileIsNotReadAndTheUploadIsSwedish(t *testing.T) {
app, sub := newUploadApp(t, audioOK, 200<<20)
rec := do(t, app, langReq(t, []string{"title", "file", "language"}, "auto"))
require.Equal(t, http.StatusSeeOther, rec.Code)
require.Len(t, sub.jobs, 1)
assert.Equal(t, domain.LangSwedish, sub.jobs[0].Language)
}
func TestUploadPost_ALanguageOutsideTheClosedSetIsRefusedBeforeAnythingIsSpooledOrStored(t *testing.T) {
for _, bad := range []string{"en", "SV", " sv", "sv,auto", "../auto", strings.Repeat("a", 5000)} {
t.Run(bad[:min(len(bad), 12)], func(t *testing.T) {
app, sub := newUploadApp(t, audioOK, 200<<20)
rec := do(t, app, langReq(t, []string{"title", "language", "file"}, bad))
assert.Equal(t, http.StatusBadRequest, rec.Code)
assert.Empty(t, sub.jobs, "a refused post must not enqueue transcription")
rows, err := newStore(t).ListUploads(context.Background(), userID)
require.NoError(t, err)
assert.Empty(t, rows, "and must leave no row")
assert.Contains(t, body(t, rec), "Choose a language from the list.")
})
}
}
func TestUploadPage_TheFormOffersSwedishPreselectedAndOtherBeforeTheFilePart(t *testing.T) {
app, _ := newUploadApp(t, audioOK, 200<<20)
b := body(t, do(t, app, get("/uploads")))
sel := strings.Index(b, `name="language"`)
file := strings.Index(b, `name="file"`)
require.True(t, sel >= 0, "the form has a language field")
assert.True(t, sel < file, "the field must come before the file part: the stream ends at the file")
assert.Regexp(t, `<option value="sv" selected`, b, "Swedish is preselected: today's behaviour")
assert.Contains(t, b, `<option value="auto"`)
}
func TestUploadPage_NamesTheLanguageInEveryState(t *testing.T) {
for _, status := range []string{"queued", "transcribing", "done", "failed"} {
for _, tt := range []struct {
language domain.Language
want string
}{
{domain.LangSwedish, "Swedish"},
{domain.LangAuto, "Other (detected by the engine)"},
} {
t.Run(status+"/"+string(tt.language), func(t *testing.T) {
app, _ := newUploadApp(t, audioOK, 200<<20)
id := seedUpload(t, userID, "m", status, "Synthetic transcript text.")
_, err := rawPool(t).Exec(context.Background(), `UPDATE uploaded_transcripts SET language = $2 WHERE id = $1`, id, string(tt.language))
require.NoError(t, err)
b := body(t, do(t, app, get("/uploads/"+id)))
assert.Contains(t, b, "Language: "+tt.want)
})
}
}
}
+3 -1
View File
@@ -13,6 +13,8 @@ import (
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"git.d-ma.be/mathias/tapir/internal/domain"
) )
// Error paths of the upload routes log a kind, never the error's text, which can carry // Error paths of the upload routes log a kind, never the error's text, which can carry
@@ -37,7 +39,7 @@ func (l *lockedBuf) String() string {
type failingUploader struct{ err error } type failingUploader struct{ err error }
func (f failingUploader) Accept(context.Context, string, string, io.Reader) (string, error) { func (f failingUploader) Accept(context.Context, string, string, domain.Language, io.Reader) (string, error) {
return "", f.err return "", f.err
} }
File diff suppressed because it is too large Load Diff
@@ -123,6 +123,10 @@ var scenarioCoverage = map[string]string{
"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", "A transcript made by the cloud fallback says the audio left the homelab": "TestUploadStatus_ACloudServedTranscriptSaysTheAudioLeftTheHomelab",
"I choose the language of a recording": "TestUploadPost_TheLanguageFieldReachesTheJobAndTheRow",
"Swedish stays the default": "TestUploadPost_NoLanguageFieldIsSwedishAsBefore",
"A language outside the list is refused": "TestUploadPost_ALanguageOutsideTheClosedSetIsRefusedBeforeAnythingIsSpooledOrStored",
"A recording in another language never reaches the Swedish-only engine": "TestEnsemble_AnAutoLanguageUploadNeverReachesTheBaseEngineAndIsNotReconciled",
"I can delete a finished recording": "TestUploadDelete_AnOwnerRemovesAFinishedOrFailedRecordingAndLandsOnTheList", "I can delete a finished recording": "TestUploadDelete_AnOwnerRemovesAFinishedOrFailedRecordingAndLandsOnTheList",
"A recording still being transcribed cannot be deleted": "TestUploadDelete_ARecordingStillBeingTranscribedIsRefusedWith409AndKept", "A recording still being transcribed cannot be deleted": "TestUploadDelete_ARecordingStillBeingTranscribedIsRefusedWith409AndKept",