Compare commits
6
Commits
25628a9994
...
c24573bd9f
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c24573bd9f | ||
|
|
fc1e8e7ba8 | ||
|
|
4a0226d8bd | ||
|
|
2e54d80a78 | ||
|
|
01c1606caa | ||
|
|
9aa01468ce |
+3
-1
@@ -1556,7 +1556,9 @@ deferred. The scope cut — chat first, no rich analysis, no video — is what k
|
||||
mechanism), confirm-then-delete on the recording's page. A recording that is **queued or
|
||||
transcribing is refused (409) until it ends**, because the worker would otherwise write to a
|
||||
row that is gone; a job whose service was lost stays undeletable until the next start's sweep
|
||||
marks it failed. The home list now links to `/uploads` when uploads are mounted.
|
||||
marks it failed. The refusal reflects the row's state when the delete was attempted: a
|
||||
recording that finishes in the same instant still gets a 409 once, and a retry deletes it.
|
||||
The home list now links to `/uploads` when uploads are mounted.
|
||||
|
||||
**Reversibility.** Additive: one table (down-migration drops it), one adapter, one upload
|
||||
surface, one image change. Removing the upload route removes the feature; `transcripts`, `videos`
|
||||
|
||||
@@ -79,3 +79,78 @@ func TestClient_AFailingBodyIsReadOnlyUpToTheErrorBound(t *testing.T) {
|
||||
t.Fatalf("the client kept reading an endless error body: took %s", time.Since(start))
|
||||
}
|
||||
}
|
||||
|
||||
// sizedGateway answers 200 with a valid completion padded with trailing whitespace to
|
||||
// exactly size bytes, so the boundary of the cap can be probed to the byte.
|
||||
func sizedGateway(t *testing.T, size int) *httptest.Server {
|
||||
t.Helper()
|
||||
const base = `{"choices":[{"message":{"content":"ok"}}]}`
|
||||
payload := base + strings.Repeat(" ", size-len(base))
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(payload))
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
return srv
|
||||
}
|
||||
|
||||
func TestClient_ABodyOfExactlyTheCapIsReadAndOneByteMoreIsRefused(t *testing.T) {
|
||||
const capBytes = 4096
|
||||
for _, tt := range []struct {
|
||||
name string
|
||||
size int
|
||||
wantErr bool
|
||||
}{
|
||||
{"one under", capBytes - 1, false},
|
||||
{"exactly the cap", capBytes, false},
|
||||
{"one over", capBytes + 1, true},
|
||||
} {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
c := New(sizedGateway(t, tt.size).URL, "", "m", 5*time.Second, WithMaxResponseBytes(capBytes))
|
||||
|
||||
got, err := c.Complete(context.Background(), "s", "u")
|
||||
|
||||
if tt.wantErr {
|
||||
if err == nil || !strings.Contains(err.Error(), "exceeds") {
|
||||
t.Fatalf("want a response-too-large error, got %q, %v", got, err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if err != nil || got != "ok" {
|
||||
t.Fatalf("want ok, got %q, %v", got, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// New is what production calls: no option. The default is therefore the live bound.
|
||||
func TestClient_TheDefaultCapIsEightMiBWhenNoOptionIsGiven(t *testing.T) {
|
||||
for _, tt := range []struct {
|
||||
name string
|
||||
size int
|
||||
wantErr bool
|
||||
}{
|
||||
{"exactly 8 MiB", 8 << 20, false},
|
||||
{"8 MiB plus one", 8<<20 + 1, true},
|
||||
} {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
c := New(sizedGateway(t, tt.size).URL, "", "m", 10*time.Second)
|
||||
|
||||
_, err := c.Complete(context.Background(), "s", "u")
|
||||
|
||||
if (err != nil) != tt.wantErr {
|
||||
t.Fatalf("size %d: err = %v, wantErr %v", tt.size, err, tt.wantErr)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_ANonPositiveCapOptionKeepsTheDefault(t *testing.T) {
|
||||
for _, n := range []int64{0, -1} {
|
||||
c := New(sizedGateway(t, 1024).URL, "", "m", 5*time.Second, WithMaxResponseBytes(n))
|
||||
|
||||
if got, err := c.Complete(context.Background(), "s", "u"); err != nil || got != "ok" {
|
||||
t.Fatalf("WithMaxResponseBytes(%d): want ok from the default cap, got %q, %v", n, got, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -202,6 +202,7 @@ func (c *Client) Complete(ctx context.Context, system, user string) (string, err
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
// Only the bounded head is ever used (HTTPError.Error truncates to it), so read no more.
|
||||
// A read error here only shortens a diagnostic; the status code below is what callers act on.
|
||||
head, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBody+1))
|
||||
return "", &HTTPError{Code: resp.StatusCode, Body: string(head)}
|
||||
}
|
||||
|
||||
@@ -57,8 +57,11 @@ func WithClock(now func() time.Time) Option { return func(t *Tracker) { t.now =
|
||||
func WithMinSave(d time.Duration) Option { return func(t *Tracker) { t.minSave = d } }
|
||||
|
||||
// WithSaveTimeout bounds one save or heartbeat write. Progress is cosmetic, so a database
|
||||
// that stops answering costs a dropped write, never a stalled job: saves run under the
|
||||
// tracker's lock, which Begin and Advance (called by the engines) need (#51).
|
||||
// that stops answering costs a dropped write rather than an unbounded stall. It is still a
|
||||
// delay: saves run under the tracker's lock, which Begin and Advance (called by the engines)
|
||||
// need, so each transition can wait up to this long, and a hung database slows a job by
|
||||
// that much per call, out of the job's own time budget (#51). A dropped final save is not
|
||||
// retried; the row keeps its last saved step.
|
||||
func WithSaveTimeout(d time.Duration) Option { return func(t *Tracker) { t.saveTimeout = d } }
|
||||
|
||||
// Tracker implements Reporter for one job.
|
||||
|
||||
@@ -96,3 +96,59 @@ func TestTracker_TheStillRunningLineIsNotHeldUpBehindAHungSave(t *testing.T) {
|
||||
|
||||
require.Eventually(t, func() bool { return strings.Contains(buf.String(), "still running") }, 3*time.Second, 5*time.Millisecond)
|
||||
}
|
||||
|
||||
// deadlineSink records how long each write was given, and answers at once.
|
||||
type deadlineSink struct {
|
||||
mu sync.Mutex
|
||||
saves []time.Duration
|
||||
beats []time.Duration
|
||||
noDLOK bool
|
||||
}
|
||||
|
||||
func (d *deadlineSink) record(ctx context.Context, into *[]time.Duration) {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
dl, ok := ctx.Deadline()
|
||||
if !ok {
|
||||
d.noDLOK = true
|
||||
return
|
||||
}
|
||||
*into = append(*into, time.Until(dl))
|
||||
}
|
||||
|
||||
func (d *deadlineSink) SaveProgress(ctx context.Context, _ []domain.ProgressStep, _ time.Time) error {
|
||||
d.record(ctx, &d.saves)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *deadlineSink) Beat(ctx context.Context, _ time.Time) error {
|
||||
d.record(ctx, &d.beats)
|
||||
return nil
|
||||
}
|
||||
|
||||
// The production value is the one the worker gets: it builds a Tracker without options.
|
||||
// A zero or tiny default would drop every write, a huge one would bound nothing.
|
||||
func TestTracker_TheDefaultSaveAndBeatDeadlineIsAFewSecondsAway(t *testing.T) {
|
||||
sink := &deadlineSink{}
|
||||
tr := progress.NewTracker(context.Background(), sink, slog.New(slog.NewTextHandler(&lockedBuf{}, nil)),
|
||||
[]progress.Planned{{Key: "a", Label: "A"}})
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
go tr.Heartbeat(ctx, 5*time.Millisecond, time.Hour)
|
||||
|
||||
tr.Begin("a", 0)
|
||||
require.Eventually(t, func() bool {
|
||||
sink.mu.Lock()
|
||||
defer sink.mu.Unlock()
|
||||
return len(sink.beats) > 0
|
||||
}, 3*time.Second, 5*time.Millisecond)
|
||||
|
||||
sink.mu.Lock()
|
||||
defer sink.mu.Unlock()
|
||||
require.False(t, sink.noDLOK, "a write was given no deadline at all")
|
||||
require.NotEmpty(t, sink.saves)
|
||||
for _, d := range append(append([]time.Duration{}, sink.saves...), sink.beats...) {
|
||||
require.Greater(t, d, time.Second, "the deadline is already (nearly) spent when the write starts")
|
||||
require.LessOrEqual(t, d, 5*time.Second, "the default bound is 5 s")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
package uploads
|
||||
|
||||
// Observes what jobSink hands the store: the heartbeat's timestamp is the one the tracker
|
||||
// stamped (its application clock), not a second reading taken here (#51).
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.d-ma.be/mathias/tapir/internal/domain"
|
||||
)
|
||||
|
||||
type atStore struct{ beatAt, saveAt time.Time }
|
||||
|
||||
func (*atStore) StartTranscribing(context.Context, string, string) error { return nil }
|
||||
func (*atStore) CompleteUpload(context.Context, string, string, domain.Transcription) error {
|
||||
return nil
|
||||
}
|
||||
func (*atStore) FailUpload(context.Context, string, string, domain.UploadErrorKind) error { return nil }
|
||||
func (s *atStore) SaveProgress(_ context.Context, _, _ string, _ []domain.ProgressStep, at time.Time) error {
|
||||
s.saveAt = at
|
||||
return nil
|
||||
}
|
||||
func (s *atStore) Heartbeat(_ context.Context, _, _ string, at time.Time) error {
|
||||
s.beatAt = at
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestJobSink_ForwardsTheTrackersTimestampToTheStoreUnchanged(t *testing.T) {
|
||||
st := &atStore{}
|
||||
sink := jobSink{st: st, userID: "u", id: "i"}
|
||||
stamp := time.Date(2020, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||
|
||||
if err := sink.Beat(context.Background(), stamp); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := sink.SaveProgress(context.Background(), nil, stamp); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if !st.beatAt.Equal(stamp) {
|
||||
t.Errorf("heartbeat reached the store stamped %v, want the tracker's %v", st.beatAt, stamp)
|
||||
}
|
||||
if !st.saveAt.Equal(stamp) {
|
||||
t.Errorf("progress save reached the store stamped %v, want the tracker's %v", st.saveAt, stamp)
|
||||
}
|
||||
}
|
||||
@@ -113,7 +113,6 @@ func TestCrossOrigin_AForeignPostIsRefusedOnEachListedStateChangingPath(t *testi
|
||||
"/v/abc/summarize",
|
||||
"/v/abc/retry-now",
|
||||
"/v/abc/chat",
|
||||
"/uploads/00000000-0000-0000-0000-000000000000/delete",
|
||||
}
|
||||
for _, path := range routes {
|
||||
t.Run(path, func(t *testing.T) {
|
||||
@@ -126,6 +125,20 @@ func TestCrossOrigin_AForeignPostIsRefusedOnEachListedStateChangingPath(t *testi
|
||||
}
|
||||
}
|
||||
|
||||
func TestCrossOrigin_ARefusedRecordingDeletePostLeavesTheRecordingInPlace(t *testing.T) {
|
||||
for name, h := range foreign {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
app, _ := newUploadApp(t, audioOK, 200<<20)
|
||||
id := seedUpload(t, userID, "m", "done", "Synthetic transcript text.")
|
||||
|
||||
rec := do(t, app, withHeaders(httptest.NewRequest(http.MethodPost, "/uploads/"+id+"/delete", nil), h))
|
||||
|
||||
assert.Equal(t, http.StatusForbidden, rec.Code)
|
||||
assert.True(t, uploadExists(t, id), "a refused post must not delete the recording")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// A login redirect comes back to /auth/callback as a cross-site GET navigation, and
|
||||
// health probes carry no headers: safe methods must never be refused on origin grounds.
|
||||
func TestCrossOrigin_ACrossSiteGETIsNotRefused(t *testing.T) {
|
||||
|
||||
+20
-2
@@ -273,12 +273,30 @@ func titleFromFilename(name string) string {
|
||||
// cloudProvider names the cloud service that produced a transcript, or false when it was
|
||||
// served locally (or by an unknown endpoint). "Local" is config.IsLocalModel: the same
|
||||
// definition the start-up validation uses, not a second string match here (#52).
|
||||
//
|
||||
// The endpoint is one alias, except for the two-engine path (ADR-034), which writes
|
||||
// "<base> + <alt>, reconciled by <model>". Every alias in it is judged on its own, so a
|
||||
// composite of local models is local and one naming a cloud alias names that provider.
|
||||
func cloudProvider(endpoint string) (string, bool) {
|
||||
if endpoint == "" || config.IsLocalModel(endpoint) {
|
||||
if strings.TrimSpace(endpoint) == "" {
|
||||
return "", false
|
||||
}
|
||||
provider, _, _ := strings.Cut(endpoint, "/")
|
||||
aliases := []string{}
|
||||
for _, tok := range strings.Fields(endpoint) {
|
||||
if tok = strings.Trim(tok, ","); strings.Contains(tok, "/") {
|
||||
aliases = append(aliases, tok)
|
||||
}
|
||||
}
|
||||
if len(aliases) == 0 {
|
||||
aliases = append(aliases, strings.TrimSpace(endpoint))
|
||||
}
|
||||
for _, a := range aliases {
|
||||
if !config.IsLocalModel(a) {
|
||||
provider, _, _ := strings.Cut(a, "/")
|
||||
return provider, true
|
||||
}
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
|
||||
// hasNote reports whether any phase carries a note. A note is the only warning for a
|
||||
|
||||
@@ -53,3 +53,32 @@ func TestUploadStatus_AnEndpointThatOnlyLooksLocalIsStillNamedAsCloud(t *testing
|
||||
|
||||
assert.Contains(t, b, "left the homelab")
|
||||
}
|
||||
|
||||
// The two-engine path (ADR-034) writes one composite endpoint string. Every name in it is
|
||||
// a local model, so the audio never left the homelab and the note must not say it did.
|
||||
func TestUploadStatus_AReconciledAllLocalEndpointCarriesNoCloudNote(t *testing.T) {
|
||||
for _, endpoint := range []string{
|
||||
"iguana/kb-whisper + koala/whisper, reconciled by iguana/gemma4-26b",
|
||||
"iguana/kb-whisper + koala/whisper, reconciled by iguana/gemma4-26b\n",
|
||||
} {
|
||||
t.Run(endpoint, func(t *testing.T) {
|
||||
app, _ := newUploadApp(t, audioOK, 200<<20)
|
||||
id := seedUpload(t, userID, "m", "done", "Synthetic transcript text.")
|
||||
setEndpoint(t, id, endpoint)
|
||||
|
||||
b := body(t, do(t, app, get("/uploads/"+id+"/status")))
|
||||
|
||||
assert.NotContains(t, b, "left the homelab")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestUploadStatus_AReconciledEndpointNamingACloudEngineStillNamesTheCloudProvider(t *testing.T) {
|
||||
app, _ := newUploadApp(t, audioOK, 200<<20)
|
||||
id := seedUpload(t, userID, "m", "done", "Synthetic transcript text.")
|
||||
setEndpoint(t, id, "iguana/kb-whisper + berget/kb-whisper, reconciled by iguana/gemma4-26b")
|
||||
|
||||
b := body(t, do(t, app, get("/uploads/"+id+"/status")))
|
||||
|
||||
assert.Contains(t, b, "cloud service (berget)")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user