Compare commits
17
Commits
sprint/v032
...
v0.33.0
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b9643f5fb7 | ||
|
|
cace7d179d | ||
|
|
550764d43c | ||
|
|
f704258fdf | ||
|
|
e3288d0654 | ||
|
|
e0c53f8719 | ||
|
|
0828f0debf | ||
|
|
7282893877 | ||
|
|
cc6fe7083d | ||
|
|
0338b18a17 | ||
|
|
ed9d76b0cf | ||
|
|
0b807c6bcd | ||
|
|
8e4872d2e7 | ||
|
|
4affeb12e8 | ||
|
|
460e07719d | ||
|
|
7b7da650d4 | ||
|
|
85f4cc3236 |
+10
-1
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,69 @@
|
|||||||
|
# ADR-034 second blind check: protocol (ready to run; tapir#33, D9, D10)
|
||||||
|
|
||||||
|
This repo is public and the recordings are private, so this file holds the method only. The
|
||||||
|
sheet, the key and the filled-in verdicts stay outside the repo, and the trial data is deleted
|
||||||
|
after the check. Results go back here as aggregate numbers, as in
|
||||||
|
`reconcile-blind-check-2026-10-02.md`.
|
||||||
|
|
||||||
|
## What the check decides
|
||||||
|
|
||||||
|
ADR-034 (two engines, a local model reconciles their disagreements) is `Proposed` and off in
|
||||||
|
production. Two conditions were set for keeping it. The first run answered the first one on one
|
||||||
|
reading only and did not answer the second:
|
||||||
|
|
||||||
|
1. **The reconciled reading is right at least 85% of the time.** Open flaw in the bar: it did not
|
||||||
|
say how ties count. The first run gave 95% (decided items only) or 70% (all 30), so it was met on
|
||||||
|
one reading and missed on the other.
|
||||||
|
2. **It is no worse than the better single engine.** Not answered: before ADR-035 the spots the
|
||||||
|
model *kept* as the base were not stored, so kb-whisper alone could not be compared.
|
||||||
|
|
||||||
|
ADR-035 now stores every judged spot (`reconciliation.contested`: both readings, which one was
|
||||||
|
used, unsure or not, a few words of left context, a time). That makes condition 2 measurable.
|
||||||
|
|
||||||
|
## Rules to fix BEFORE looking at any result (the owner confirms or changes them)
|
||||||
|
|
||||||
|
| Rule | Proposed |
|
||||||
|
|---|---|
|
||||||
|
| Who judges | The speaker, hearing the audio where the text cannot settle it (D9). Not Opus: it cannot hear, and its agreement with the speaker was only shown on items picked as decidable. |
|
||||||
|
| Sample | 30 spots drawn at random with a fixed seed, from ALL judged spots (changed and kept-base alike), not only the changed ones. Insertions and deletions that cannot be placed are listed separately and not sampled. |
|
||||||
|
| Verdict per spot | `x`, `y` (the two readings, shown in random order, key kept apart), `neither`, or `cannot tell`. |
|
||||||
|
| Condition 1 | Reconciled is right on at least 85% of the spots where the judge could tell (`cannot tell` excluded, reported, and `neither` counts as not right). Also report all-30. Pass needs the decided-only figure at 85% or more AND the all-30 figure above 70%. |
|
||||||
|
| Condition 2 | Over the same spots, the reconciled text is right at least as often as the better single engine (base engine alone is right where the base reading is right; the other engine alone where the alt reading is right). Ties favour the single engine. |
|
||||||
|
| If it fails | Switch reconcile off by configuration; the single-engine path is unchanged (ADR-034 reversibility). |
|
||||||
|
|
||||||
|
Why a fixed all-30 floor: the literal 70% last time sat well under the bar, and "decided only" can
|
||||||
|
hide a model that mostly lands on spelling differences.
|
||||||
|
|
||||||
|
## What a run needs (D10: the owner is present)
|
||||||
|
|
||||||
|
- The same real recording as before (about 60 minutes). Only the owner has it.
|
||||||
|
- A **throwaway** tapir instance with its own Postgres, not the production pod, so production's
|
||||||
|
key pin and deployment are untouched. Settings: `TAPIR_RECONCILE_BASE_MODEL=iguana/pianissimo`,
|
||||||
|
`TAPIR_TRANSCRIBE_MODEL=iguana/kb-whisper`, `TAPIR_RECONCILE_MODEL=iguana/gemma4-31b`, cloud
|
||||||
|
fallbacks empty. Job time was 21 min 49 s per 63.6 min last run.
|
||||||
|
- iguana with Low Power Mode off and gemma4-31b able to load (the production-iguana model load
|
||||||
|
that needs a quiet window; D6).
|
||||||
|
|
||||||
|
## Steps
|
||||||
|
|
||||||
|
1. Start the throwaway instance, upload the recording, wait for `done` (about 22 minutes).
|
||||||
|
2. Read `reconciliation.contested` from the throwaway database. From it build, **outside the repo**:
|
||||||
|
`sheet` (spot number, time, left context, reading x, reading y, random order) and `key` (which
|
||||||
|
of x/y is the base, which is the other engine, which one the reconciler used).
|
||||||
|
3. The speaker fills in the sheet (about 20 to 30 minutes for 30 spots, audio open at the times).
|
||||||
|
The agent does not see the key until the sheet is returned.
|
||||||
|
4. Score with the rules above. Write the aggregate numbers into a dated note next to this one,
|
||||||
|
and amend ADR-034's validation section (kept, or switched off, and the tie rule used).
|
||||||
|
5. Delete the throwaway instance, its database and the sheet, as last time.
|
||||||
|
6. Only if the check passes and the owner wants it in production: enabling reconcile there is
|
||||||
|
the separate D10 step (it needs `iguana/pianissimo` and the reconciler added to the tapir key
|
||||||
|
pin in the infra repo plus the env vars on the deployment, `risk/medium`, owner present).
|
||||||
|
|
||||||
|
## Open and owner-only
|
||||||
|
|
||||||
|
- Confirm or change the rules table above before the run.
|
||||||
|
- The **tool that builds the sheet and key** is the one piece not in the repo (last time it was a
|
||||||
|
one-off script). It can be built test-first on request; until then the agent builds it ad hoc
|
||||||
|
outside the repo.
|
||||||
|
- A Berget adjudicator (D8, tapir#45) is a different question with its own measurement and is
|
||||||
|
not part of this check: this check judges the local setup that ADR-034 describes.
|
||||||
@@ -1,7 +1,8 @@
|
|||||||
# Sprint: audio upload follow-ups (Sweep Sprint, 2026-10)
|
# Sprint: audio upload follow-ups (Sweep Sprint, 2026-10)
|
||||||
|
|
||||||
Method: Sweep Sprint (phase 1 parallel read-only research per issue: **done**; phase 2 sequential
|
Method: Sweep Sprint (phase 1 parallel read-only research per issue: **done**; phase 2 sequential
|
||||||
execution, one issue per commit: **not started, waits for the owner's go**). Closing oath below.
|
execution, one issue per commit: **v0.31.1 and v0.32.0 shipped 2026-10-03/04, the owner-held items
|
||||||
|
below remain open**). Closing oath below.
|
||||||
|
|
||||||
Milestones: [v0.31.1 - security and robustness](https://git.d-ma.be/mathias/tapir/milestone/10),
|
Milestones: [v0.31.1 - security and robustness](https://git.d-ma.be/mathias/tapir/milestone/10),
|
||||||
[v0.32.0 - audio upload follow-ups](https://git.d-ma.be/mathias/tapir/milestone/11).
|
[v0.32.0 - audio upload follow-ups](https://git.d-ma.be/mathias/tapir/milestone/11).
|
||||||
@@ -63,8 +64,8 @@ Release v0.31.1: one adversarial review round over 2-5, then tag, push, observe.
|
|||||||
| D5 | #38: Berget length measurement needs a throwaway key | Yes, throwaway key | #38 |
|
| D5 | #38: Berget length measurement needs a throwaway key | Yes, throwaway key | #38 |
|
||||||
| D6 | #38: load large models on production iguana | Yes, in a quiet window with the owner present | #38 |
|
| D6 | #38: load large models on production iguana | Yes, in a quiet window with the owner present | #38 |
|
||||||
| D7 | #46: where the ffmpeg tarball lives | Do nothing now; file an issue | none |
|
| D7 | #46: where the ffmpeg tarball lives | Do nothing now; file an issue | none |
|
||||||
| D8 | #45: evaluate Berget as the reconciler | No; stay local-only | #45 |
|
| D8 | #45: evaluate Berget as the reconciler | **Decided 2026-10-04 by the owner: yes, Berget may be used, and each user must be able to keep a recording on local models only** (reverses the recommendation). Opt-out first: #60; #45 stays as the model measurement | #60, #45 |
|
||||||
| D9 | Who judges ADR-034's second blind check | The owner; do not reuse the one-off Opus authorisation | ADR-034 |
|
| D9 | Who judges ADR-034's second blind check | The owner; do not reuse the one-off Opus authorisation. Protocol: `docs/research/reconcile-second-check-protocol.md` | ADR-034 |
|
||||||
| D10 | When to enable reconcile in production | Needs `iguana/pianissimo` in the key pin and env vars; owner present | ADR-034 |
|
| D10 | When to enable reconcile in production | Needs `iguana/pianissimo` in the key pin and env vars; owner present | ADR-034 |
|
||||||
|
|
||||||
## Dropped or reduced, with reasons
|
## Dropped or reduced, with reasons
|
||||||
@@ -86,3 +87,29 @@ Release v0.31.1: one adversarial review round over 2-5, then tag, push, observe.
|
|||||||
SemVer: v0.31.1 patch (fixes, no new behaviour); v0.32.0 minor (nav link, delete, cloud labelling).
|
SemVer: v0.31.1 patch (fixes, no new behaviour); v0.32.0 minor (nav link, delete, cloud labelling).
|
||||||
Every push to tapir main redeploys the pod and fails any job in flight: push when no upload is
|
Every push to tapir main redeploys the pod and fails any job in flight: push when no upload is
|
||||||
running (check the pod log first).
|
running (check the pod log first).
|
||||||
|
|
||||||
|
## Status (2026-10-04)
|
||||||
|
|
||||||
|
**Shipped and deployed:** v0.31.1 (PR #55, merge e037c6b) and v0.32.0 (PR #58, merge 85f4cc3; pod
|
||||||
|
tapir-597dbfccc-ng78q 1/1, 0 restarts, image 85f4cc3). Each went through an adversarial review
|
||||||
|
round by agent with the verdicts recorded on the PR (v0.31.1: BLOCK then APPROVE; v0.32.0: BLOCK,
|
||||||
|
BLOCK, APPROVE, APPROVE; every block was a real defect, fixed test-first).
|
||||||
|
|
||||||
|
**Live evidence:** a foreign-origin `POST` to `/account/delete` and to `/uploads/{id}/delete` is
|
||||||
|
refused (403) on production, a same-origin one reaches the login redirect (302).
|
||||||
|
|
||||||
|
**Not observed live (needs an authenticated session, so it is the owner's look; closed on word or screenshot):**
|
||||||
|
|
||||||
|
- #52, cloud side: a transcript served by the Berget fallback names `berget`; a reconciled all-local
|
||||||
|
one shows no note. The local side was observed on 2026-10-04 (a new clip served by iguana: chip
|
||||||
|
`iguana/…`, no cloud note). The cloud side is carried as an acceptance item in #60.
|
||||||
|
|
||||||
|
**Observed live since:** #37 (owner, 2026-10-04: link, confirm-then-delete, no control while transcribing; closed). #50: at the 2026-10-04 rollout the old pod went `Terminating` then `Completed`
|
||||||
|
(not `Error`), 0 restarts; #50 closed. No job was in flight, so a drained job is still unobserved
|
||||||
|
(a recorded limitation, D2).
|
||||||
|
|
||||||
|
**Still open (owner's call, milestone 11):** #36, #38, #45 (now the Berget adjudicator measurement), #60 (per-recording local-only choice, 4 sub-decisions), #46 remainder, #56 (slow-drip bodies hold
|
||||||
|
a receive slot), #57 (shutdown logging, second signal). Decisions D4-D6, D9, D10 above are unchanged; D8 is decided (above).
|
||||||
|
|
||||||
|
**Deferred and filed from the reviews:** #56, #57, #59 (start-up exits when the identity provider's OIDC
|
||||||
|
discovery returns 503 (seen once at 19:40 on 2026-10-03; the pod restarted itself and recovered)).
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -33,9 +33,9 @@ require (
|
|||||||
github.com/rogpeppe/go-internal v1.15.0 // indirect
|
github.com/rogpeppe/go-internal v1.15.0 // indirect
|
||||||
github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect
|
github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect
|
||||||
go.yaml.in/yaml/v2 v2.4.2 // indirect
|
go.yaml.in/yaml/v2 v2.4.2 // indirect
|
||||||
golang.org/x/sync v0.18.0 // indirect
|
golang.org/x/sync v0.21.0 // indirect
|
||||||
golang.org/x/sys v0.41.0 // indirect
|
golang.org/x/sys v0.41.0 // indirect
|
||||||
golang.org/x/text v0.31.0 // indirect
|
golang.org/x/text v0.39.0 // indirect
|
||||||
google.golang.org/protobuf v1.36.8 // indirect
|
google.golang.org/protobuf v1.36.8 // indirect
|
||||||
gopkg.in/yaml.v3 v3.0.1 // indirect
|
gopkg.in/yaml.v3 v3.0.1 // indirect
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -113,12 +113,12 @@ go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
|
|||||||
go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
|
go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
|
||||||
golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
|
golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
|
||||||
golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
|
golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
|
||||||
golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I=
|
golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM=
|
||||||
golang.org/x/sync v0.18.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
|
golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
|
||||||
golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k=
|
golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k=
|
||||||
golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
|
golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
|
||||||
golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM=
|
golang.org/x/text v0.39.0 h1:UbZz4pLOvn600D6Oh6GGEI6VAmndrEBLv8/6BEXzyus=
|
||||||
golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM=
|
golang.org/x/text v0.39.0/go.mod h1:3UwRclnC2g0TU9x8PZiyfOajCd1zaUNHF9cvqcQZ+ZM=
|
||||||
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
|
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
|
||||||
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
|
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
|
||||||
google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc=
|
google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc=
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
|||||||
@@ -32,15 +32,25 @@ 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{
|
|
||||||
|
// PlanFor implements uploads.LanguagePlanner. A recording that is not Swedish never runs the
|
||||||
|
// base engine, the comparison or the decisions (transcribeOther), so the page lists only
|
||||||
|
// what can run: the other engine, then the fallback if one is configured.
|
||||||
|
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_a", Label: "Transcribing with " + e.base.Name()},
|
||||||
{Key: "engine_b", Label: "Transcribing with " + e.alt.Name()},
|
{Key: "engine_b", Label: "Transcribing with " + e.alt.Name()},
|
||||||
{Key: "compare", Label: "Comparing the two transcripts"},
|
{Key: "compare", Label: "Comparing the two transcripts"},
|
||||||
{Key: "decide", Label: "Deciding the differences"},
|
{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")
|
||||||
|
}
|
||||||
|
|||||||
@@ -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")
|
||||||
|
}
|
||||||
|
|||||||
@@ -246,6 +246,9 @@ 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
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -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))
|
||||||
|
}
|
||||||
@@ -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.
|
||||||
|
|||||||
@@ -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)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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")
|
||||||
|
}
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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()))
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -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)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -0,0 +1,61 @@
|
|||||||
|
package oidc_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
|
||||||
|
"git.d-ma.be/mathias/tapir/internal/web/oidc"
|
||||||
|
)
|
||||||
|
|
||||||
|
func testConfig(issuer string) oidc.Config {
|
||||||
|
return oidc.Config{
|
||||||
|
Issuer: issuer,
|
||||||
|
ClientID: testClientID,
|
||||||
|
ClientSecret: "test-client-secret",
|
||||||
|
RedirectURL: "http://tapir.test/auth/callback",
|
||||||
|
SessionSecret: "test-session-secret-please-change",
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// #59: a discovery 503 at start-up is retried instead of failing New.
|
||||||
|
func TestNewRetriesDiscoveryAfter503UntilIssuerAnswers(t *testing.T) {
|
||||||
|
f := newFakeIssuer(t)
|
||||||
|
f.unavailableFor = 2
|
||||||
|
|
||||||
|
auth, err := oidc.New(context.Background(), testConfig(f.server.URL),
|
||||||
|
oidc.WithDiscoveryRetry(5, time.Millisecond))
|
||||||
|
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.NotNil(t, auth)
|
||||||
|
require.Equal(t, int32(3), f.discoveryHits.Load())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNewReturnsDiscoveryErrorAfterAttemptsExhausted(t *testing.T) {
|
||||||
|
f := newFakeIssuer(t)
|
||||||
|
f.unavailableFor = 1000
|
||||||
|
|
||||||
|
_, err := oidc.New(context.Background(), testConfig(f.server.URL),
|
||||||
|
oidc.WithDiscoveryRetry(3, time.Millisecond))
|
||||||
|
|
||||||
|
require.ErrorContains(t, err, "503")
|
||||||
|
require.Equal(t, int32(3), f.discoveryHits.Load())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNewStopsDiscoveryBackoffWhenContextCancelled(t *testing.T) {
|
||||||
|
f := newFakeIssuer(t)
|
||||||
|
f.unavailableFor = 1000
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
start := time.Now()
|
||||||
|
_, err := oidc.New(ctx, testConfig(f.server.URL),
|
||||||
|
oidc.WithDiscoveryRetry(5, time.Hour))
|
||||||
|
|
||||||
|
require.True(t, errors.Is(err, context.DeadlineExceeded), "err = %v", err)
|
||||||
|
require.Less(t, time.Since(start), 5*time.Second)
|
||||||
|
require.Equal(t, int32(1), f.discoveryHits.Load())
|
||||||
|
}
|
||||||
+66
-16
@@ -57,6 +57,10 @@ const (
|
|||||||
pendingTTL = 10 * time.Minute
|
pendingTTL = 10 * time.Minute
|
||||||
sessionCookie = "tapir_session"
|
sessionCookie = "tapir_session"
|
||||||
loginPath = "/auth/login"
|
loginPath = "/auth/login"
|
||||||
|
|
||||||
|
// Discovery retry (#59): 5 tries, 2+4+8+16 s back-off ≈ 30 s, then fail.
|
||||||
|
defaultDiscoveryAttempts = 5
|
||||||
|
defaultDiscoveryDelay = 2 * time.Second
|
||||||
)
|
)
|
||||||
|
|
||||||
// DexAuth implements web.Auth against Dex.
|
// DexAuth implements web.Auth against Dex.
|
||||||
@@ -70,6 +74,9 @@ type DexAuth struct {
|
|||||||
sessionTTL time.Duration
|
sessionTTL time.Duration
|
||||||
now func() time.Time
|
now func() time.Time
|
||||||
insecure bool // tests only: drop the Secure attribute so cookies flow over http
|
insecure bool // tests only: drop the Secure attribute so cookies flow over http
|
||||||
|
|
||||||
|
discoveryAttempts int
|
||||||
|
discoveryDelay time.Duration
|
||||||
}
|
}
|
||||||
|
|
||||||
var _ web.Auth = (*DexAuth)(nil)
|
var _ web.Auth = (*DexAuth)(nil)
|
||||||
@@ -96,14 +103,55 @@ func WithSessionTTL(ttl time.Duration) Option {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// WithDiscoveryRetry overrides the discovery attempt count and the initial
|
||||||
|
// back-off delay (doubled after each failed attempt).
|
||||||
|
func WithDiscoveryRetry(attempts int, delay time.Duration) Option {
|
||||||
|
return func(d *DexAuth) {
|
||||||
|
if attempts > 0 {
|
||||||
|
d.discoveryAttempts = attempts
|
||||||
|
}
|
||||||
|
if delay > 0 {
|
||||||
|
d.discoveryDelay = delay
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// WithInsecureCookies drops the Secure cookie attribute. TEST ONLY — it lets
|
// WithInsecureCookies drops the Secure cookie attribute. TEST ONLY — it lets
|
||||||
// the session cookie travel over plain http; never use it in production.
|
// the session cookie travel over plain http; never use it in production.
|
||||||
func WithInsecureCookies() Option {
|
func WithInsecureCookies() Option {
|
||||||
return func(d *DexAuth) { d.insecure = true }
|
return func(d *DexAuth) { d.insecure = true }
|
||||||
}
|
}
|
||||||
|
|
||||||
// New discovers the issuer and returns a configured DexAuth. ctx bounds the
|
// discover runs OIDC discovery up to attempts times, doubling delay between
|
||||||
// discovery request only.
|
// tries (#59: an issuer briefly unavailable behind its proxy answered 503 and
|
||||||
|
// serve exited). A cancelled ctx ends the back-off immediately with ctx's error.
|
||||||
|
func discover(ctx context.Context, issuer string, attempts int, delay time.Duration) (*oidc.Provider, error) {
|
||||||
|
var lastErr error
|
||||||
|
for i := range attempts {
|
||||||
|
if i > 0 {
|
||||||
|
t := time.NewTimer(delay)
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
t.Stop()
|
||||||
|
return nil, fmt.Errorf("after %d attempts (last: %v): %w", i, lastErr, ctx.Err())
|
||||||
|
case <-t.C:
|
||||||
|
}
|
||||||
|
delay *= 2
|
||||||
|
}
|
||||||
|
p, err := oidc.NewProvider(ctx, issuer)
|
||||||
|
if err == nil {
|
||||||
|
return p, nil
|
||||||
|
}
|
||||||
|
lastErr = err
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
return nil, fmt.Errorf("after %d attempts: %w", i+1, ctx.Err())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil, fmt.Errorf("after %d attempts: %w", attempts, lastErr)
|
||||||
|
}
|
||||||
|
|
||||||
|
// New discovers the issuer and returns a configured DexAuth. ctx bounds discovery,
|
||||||
|
// including its retry back-off.
|
||||||
func New(ctx context.Context, cfg Config, opts ...Option) (*DexAuth, error) {
|
func New(ctx context.Context, cfg Config, opts ...Option) (*DexAuth, error) {
|
||||||
for name, val := range map[string]string{
|
for name, val := range map[string]string{
|
||||||
"issuer": cfg.Issuer,
|
"issuer": cfg.Issuer,
|
||||||
@@ -117,28 +165,30 @@ func New(ctx context.Context, cfg Config, opts ...Option) (*DexAuth, error) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
provider, err := oidc.NewProvider(ctx, cfg.Issuer)
|
d := &DexAuth{
|
||||||
|
cfg: cfg,
|
||||||
|
pending: newPendingStore(),
|
||||||
|
secret: []byte(cfg.SessionSecret),
|
||||||
|
sessionTTL: defaultSessionTTL,
|
||||||
|
now: time.Now,
|
||||||
|
discoveryAttempts: defaultDiscoveryAttempts,
|
||||||
|
discoveryDelay: defaultDiscoveryDelay,
|
||||||
|
}
|
||||||
|
for _, o := range opts {
|
||||||
|
o(d)
|
||||||
|
}
|
||||||
|
|
||||||
|
provider, err := discover(ctx, cfg.Issuer, d.discoveryAttempts, d.discoveryDelay)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("oidc: discover issuer: %w", err)
|
return nil, fmt.Errorf("oidc: discover issuer: %w", err)
|
||||||
}
|
}
|
||||||
|
d.verifier = provider.Verifier(&oidc.Config{ClientID: cfg.ClientID})
|
||||||
d := &DexAuth{
|
d.oauth = &oauth2.Config{
|
||||||
cfg: cfg,
|
|
||||||
verifier: provider.Verifier(&oidc.Config{ClientID: cfg.ClientID}),
|
|
||||||
oauth: &oauth2.Config{
|
|
||||||
ClientID: cfg.ClientID,
|
ClientID: cfg.ClientID,
|
||||||
ClientSecret: cfg.ClientSecret,
|
ClientSecret: cfg.ClientSecret,
|
||||||
Endpoint: provider.Endpoint(),
|
Endpoint: provider.Endpoint(),
|
||||||
RedirectURL: cfg.RedirectURL,
|
RedirectURL: cfg.RedirectURL,
|
||||||
Scopes: []string{oidc.ScopeOpenID, "profile", "email"},
|
Scopes: []string{oidc.ScopeOpenID, "profile", "email"},
|
||||||
},
|
|
||||||
pending: newPendingStore(),
|
|
||||||
secret: []byte(cfg.SessionSecret),
|
|
||||||
sessionTTL: defaultSessionTTL,
|
|
||||||
now: time.Now,
|
|
||||||
}
|
|
||||||
for _, o := range opts {
|
|
||||||
o(d)
|
|
||||||
}
|
}
|
||||||
return d, nil
|
return d, nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import (
|
|||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"net/url"
|
"net/url"
|
||||||
|
"sync/atomic"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -34,6 +35,11 @@ type fakeIssuer struct {
|
|||||||
sub string
|
sub string
|
||||||
email string
|
email string
|
||||||
nonce string
|
nonce string
|
||||||
|
|
||||||
|
// discoveryHits counts requests to the discovery document; the first
|
||||||
|
// unavailableFor of them answer 503 (issuer briefly down behind its proxy).
|
||||||
|
discoveryHits atomic.Int32
|
||||||
|
unavailableFor int32
|
||||||
}
|
}
|
||||||
|
|
||||||
const testKID = "test-key"
|
const testKID = "test-key"
|
||||||
@@ -47,6 +53,10 @@ func newFakeIssuer(t *testing.T) *fakeIssuer {
|
|||||||
|
|
||||||
mux := http.NewServeMux()
|
mux := http.NewServeMux()
|
||||||
mux.HandleFunc("/.well-known/openid-configuration", func(w http.ResponseWriter, _ *http.Request) {
|
mux.HandleFunc("/.well-known/openid-configuration", func(w http.ResponseWriter, _ *http.Request) {
|
||||||
|
if f.discoveryHits.Add(1) <= f.unavailableFor {
|
||||||
|
http.Error(w, "no available server", http.StatusServiceUnavailable)
|
||||||
|
return
|
||||||
|
}
|
||||||
writeJSON(t, w, map[string]any{
|
writeJSON(t, w, map[string]any{
|
||||||
"issuer": f.server.URL,
|
"issuer": f.server.URL,
|
||||||
"authorization_endpoint": f.server.URL + "/authorize",
|
"authorization_endpoint": f.server.URL + "/authorize",
|
||||||
|
|||||||
+31
-2
@@ -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 {
|
||||||
|
|||||||
@@ -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)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+354
-341
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",
|
||||||
|
|||||||
Reference in New Issue
Block a user