mirror of
https://github.com/hanzoai/attendant.git
synced 2026-09-30 20:43:25 +00:00
a turn opens on its first word, not a second into it
Three things the live room found that the replay could not. A backchannel takes the floor when the wait is one observer interval: a 1.5s "mhm, yeah" cut a speaker's sentence in half and "reporting green" went with it, because a mutter holds a steady level while a speaker between words does not, so the room ranks the quieter one first. The wait is two seconds now. The wait runs from when a rival STARTED making sound rather than from the report that named them, or they pay for our latency twice. cleo began while ben was finishing; by the time the room stopped ranking him she was a second in, and her notes opened "owe you the capacity numbers". Held audio goes to a turn in one piece. It is 2.5s arriving at once, which is no rate a room delivers at, and a turn that queues audio so the room never waits dropped the overflow — the front of the turn, which is what holding it was for. Also: talk-over that its speaker then wins the floor with is not counted as lost, audio reaches one turn and never two, and the scribe's chunking, its rollover at the service's 600s limit, its ordering and its dropped-frame count are tested against a server that answers the way hanzoai/speech answers. Three consecutive runs against the live SFU and hanzoai/speech: 19.4s published, 17.6s transcribed, three turns, no frames lost, every sentence whole and under the right name. Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
Hanzo Dev
@@ -23,7 +23,10 @@ cost of transcribing a meeting independent of how many people are in it.
|
||||
**A turn begins before you know whose it is.** LiveKit reports a speaker on an
|
||||
interval and only when the ranking CHANGES, so the first syllables of every turn
|
||||
are spoken before any word arrives about them. Each speaker's last `lead` of
|
||||
audio is kept and handed to their turn when they take the floor.
|
||||
audio is kept and handed to their turn when they take the floor — in one piece,
|
||||
because `lead` seconds arriving at once is not a rate any room delivers at, and
|
||||
a turn that queues audio so the room never waits drops what it cannot hold.
|
||||
Frame by frame, the first thing dropped is the first thing said.
|
||||
|
||||
**A ranking is not a census.** Measured against the live SFU: a speaker stayed
|
||||
ranked first for thirteen seconds after her last packet, and five seconds passed
|
||||
@@ -32,14 +35,26 @@ that list is wrong in both directions. So the room says who is LOUDEST, the
|
||||
tracks say who is still there, and the floor reads both on its own clock.
|
||||
|
||||
**A "mhm" is not an interruption.** A rival must be ranked first for `grab`
|
||||
before the floor moves. Because `lead` covers `notice + grab`, waiting costs
|
||||
nothing: whatever they said while we waited is still in hand when they win.
|
||||
before the floor moves. One observer interval is not enough: in a live room a
|
||||
1.5 s "mhm, yeah" took the floor off a speaker mid-sentence, because a
|
||||
backchannel holds a steady level while a speaker between words does not, and
|
||||
the room ranked the quieter one first. `grab` is two seconds — longer than
|
||||
people say "yeah" for. Waiting costs nothing, because `lead` covers
|
||||
`notice + grab`: whatever the rival said while we waited is still in hand when
|
||||
they win.
|
||||
|
||||
**Nobody pays for our latency twice.** The wait runs from when a rival started
|
||||
making sound, not from the report that named them. Measured live: cleo began
|
||||
while ben was finishing, and by the time the room stopped ranking him she had
|
||||
been talking nearly a second — a wait starting there put the front of "I still
|
||||
owe you" outside `lead`, and her notes began "owe you the capacity numbers".
|
||||
|
||||
**Two people talking is one transcript.** The louder holds the floor; the other
|
||||
is counted, not transcribed — `Tally.Over` is how much. An interjection that
|
||||
matters becomes the floor within `grab` and its onset is recovered from `lead`,
|
||||
so real turn-taking survives and only backchannel is dropped. A second stream
|
||||
would put the headcount back into the bill.
|
||||
is counted, not transcribed — `Tally.Over` is how much, and it gives audio back
|
||||
when the interrupter wins the floor and their turn receives it after all, so the
|
||||
number is the loss and not the overlap. An interjection that matters becomes the
|
||||
floor and its onset comes out of `lead`, so real turn-taking survives and only
|
||||
backchannel is dropped. A second stream would put the headcount back in the bill.
|
||||
|
||||
## Measured
|
||||
|
||||
|
||||
@@ -32,14 +32,23 @@ const (
|
||||
// first half-second precedes any word we get about them.
|
||||
notice = 500 * time.Millisecond
|
||||
|
||||
// How long a rival must be the loudest before the floor moves. Long enough
|
||||
// that a backchannel does not interrupt a sentence, short enough that a real
|
||||
// interruption is transcribed as one.
|
||||
grab = 500 * time.Millisecond
|
||||
// How long a rival must be the loudest before the floor moves.
|
||||
//
|
||||
// This is the whole of what separates an interruption from an agreement, so
|
||||
// it has to outlast an agreement. It does not work at one observer interval:
|
||||
// a 1.5 s "mhm, yeah" took the floor off a speaker mid-sentence in a live
|
||||
// room, because a backchannel holds a steady level while a speaker between
|
||||
// words does not, and the room ranked the quieter one first. Two seconds is
|
||||
// longer than people say "yeah" for and shorter than they interrupt for.
|
||||
grab = 2 * time.Second
|
||||
|
||||
// How much of each speaker's audio is kept back, so that a turn can start
|
||||
// before we knew it had. Must cover notice + grab or turns lose their onset.
|
||||
lead = 1200 * time.Millisecond
|
||||
//
|
||||
// Waiting `grab` costs nothing because of this, and this costs nothing
|
||||
// either: what it holds is the rival's own speech from while we waited,
|
||||
// which was going nowhere until they won.
|
||||
lead = 2500 * time.Millisecond
|
||||
|
||||
// How long the holder can go unreported before the floor is free. Shorter
|
||||
// than this and a breath ends the turn; longer and a transcript sits open
|
||||
@@ -56,7 +65,9 @@ const (
|
||||
// feeds it, and closes it when the floor moves on. Turns never overlap, so an
|
||||
// implementation never multiplexes.
|
||||
type Turn interface {
|
||||
// Hear takes one decoded frame, in arrival order.
|
||||
// Hear takes audio, in arrival order: a frame as the room delivers it, or
|
||||
// the whole of what a speaker said before the floor was theirs, which
|
||||
// arrives in one piece when the turn opens.
|
||||
//
|
||||
// It must not block on the network. Audio arrives on a real-time clock and
|
||||
// every speaker's track is behind this call, so a stall here is lost audio,
|
||||
@@ -85,10 +96,19 @@ type Tally struct {
|
||||
Faults int // scribe calls that failed
|
||||
}
|
||||
|
||||
// clip is one frame and when it arrived.
|
||||
// clip is one frame, when it arrived, whether it was counted against the holder
|
||||
// at the time, and whether a turn has had it.
|
||||
//
|
||||
// A frame spoken over the holder is a loss until its speaker takes the floor and
|
||||
// it is handed to their turn after all, so the count has to be able to give it
|
||||
// back. And a speaker who takes the floor, loses it and takes it again inside
|
||||
// `lead` still has those frames in hand: sent twice, the meeting notes say what
|
||||
// they said twice.
|
||||
type clip struct {
|
||||
at time.Time
|
||||
pcm []int16
|
||||
at time.Time
|
||||
pcm []int16
|
||||
over bool
|
||||
sent bool
|
||||
}
|
||||
|
||||
// Floor awards the meeting's floor and gives the holder's audio to a scribe.
|
||||
@@ -172,6 +192,7 @@ func (f *Floor) Hear(speaker string, pcm []int16) {
|
||||
if f.turn != nil {
|
||||
f.turn.Hear(held.pcm)
|
||||
f.tally.Kept += len(pcm)
|
||||
f.keep[speaker][len(f.keep[speaker])-1].sent = true
|
||||
}
|
||||
return
|
||||
}
|
||||
@@ -179,6 +200,7 @@ func (f *Floor) Hear(speaker string, pcm []int16) {
|
||||
// not transcribed — counted here so the loss is a number and not a guess.
|
||||
if f.loud[speaker] {
|
||||
f.tally.Over += len(pcm)
|
||||
f.keep[speaker][len(f.keep[speaker])-1].over = true
|
||||
}
|
||||
}
|
||||
|
||||
@@ -263,12 +285,32 @@ func (f *Floor) contest(loud []string, now time.Time) (string, bool) {
|
||||
return top, true // an open floor is taken at once; there is nothing to protect
|
||||
}
|
||||
if f.rival != top {
|
||||
f.rival, f.from = top, now
|
||||
return "", false
|
||||
f.rival, f.from = top, f.began(top, now)
|
||||
}
|
||||
return top, now.Sub(f.from) >= grab
|
||||
}
|
||||
|
||||
// began is when the rival's audio starts in what is still held — how long they
|
||||
// have been talking, as nearly as anything here can know it.
|
||||
//
|
||||
// The wait runs from THERE and not from the report that named them, or a rival
|
||||
// pays for our latency twice: once waiting to be noticed and again waiting to be
|
||||
// believed. Measured live, cleo began talking while ben was finishing, and by
|
||||
// the time the room stopped ranking him she had been going nearly a second — a
|
||||
// wait starting then put the front of "I still owe you" outside `lead`, and her
|
||||
// notes began "owe you the capacity numbers".
|
||||
func (f *Floor) began(who string, now time.Time) time.Time {
|
||||
clips := since(f.keep[who], now.Add(-lead))
|
||||
at := now
|
||||
for i := len(clips) - 1; i >= 0; i-- {
|
||||
if at.Sub(clips[i].at) > notice {
|
||||
break // a gap this long: what came before it was a different stretch
|
||||
}
|
||||
at = clips[i].at
|
||||
}
|
||||
return at
|
||||
}
|
||||
|
||||
// award moves the floor to who: opens their turn, hands it the audio they have
|
||||
// already spoken, and closes the turn that was open.
|
||||
//
|
||||
@@ -291,10 +333,29 @@ func (f *Floor) award(who string, now time.Time) {
|
||||
default:
|
||||
f.turn, f.holder = turn, who
|
||||
f.tally.Turns++
|
||||
for _, c := range since(f.keep[who], now.Add(-lead)) {
|
||||
turn.Hear(c.pcm)
|
||||
f.tally.Kept += len(c.pcm)
|
||||
f.tally.Lead += len(c.pcm)
|
||||
|
||||
// What they had already said, in ONE piece. It is `lead` seconds arriving
|
||||
// at once, which is nothing like the pace a room delivers at: handed over
|
||||
// frame by frame it is a hundred and twenty-five of them in a moment, and
|
||||
// a turn that queues audio so the room never waits drops the overflow —
|
||||
// which is the onset, the very thing this exists to protect. Measured
|
||||
// live: cleo's turn began "owe you the capacity numbers".
|
||||
var began []int16
|
||||
held := since(f.keep[who], now.Add(-lead))
|
||||
for i := range held {
|
||||
if held[i].sent {
|
||||
continue
|
||||
}
|
||||
held[i].sent = true
|
||||
began = append(began, held[i].pcm...)
|
||||
if held[i].over {
|
||||
f.tally.Over -= len(held[i].pcm) // spoken over the last holder, kept anyway
|
||||
}
|
||||
}
|
||||
if len(began) > 0 {
|
||||
turn.Hear(began)
|
||||
f.tally.Kept += len(began)
|
||||
f.tally.Lead += len(began)
|
||||
}
|
||||
}
|
||||
f.mu.Unlock()
|
||||
|
||||
+105
-9
@@ -201,17 +201,19 @@ func TestBackchannelDoesNotTakeTheFloor(t *testing.T) {
|
||||
f.Speaking([]string{"ana"})
|
||||
|
||||
// ana keeps talking. ben mutters over her, loud enough that the room ranks
|
||||
// him first and long enough that it says so twice — a "yeah" outlasts an
|
||||
// observer interval — but not as long as grab.
|
||||
// him first — "Mhm. Yeah." runs about a second and a half — long enough that the room
|
||||
// ranks him first several times over, which is exactly the case that took
|
||||
// the floor off a speaker in a live room when grab was one interval.
|
||||
both := []voice{{"ana", 1}, {"ben", 2}}
|
||||
talk(f, c, 100*time.Millisecond, both...) // ben starts; the room has not noticed
|
||||
f.Speaking([]string{"ben", "ana"}) // now it ranks him first
|
||||
talk(f, c, 200*time.Millisecond, both...)
|
||||
f.Speaking([]string{"ben", "ana"}) // and still does, 200ms later
|
||||
if got := f.Holder(); got != "ana" {
|
||||
t.Fatalf("floor went to %q after 200ms; a rival owes %v", got, grab)
|
||||
const mhm = 1500 * time.Millisecond
|
||||
for spent := time.Duration(0); spent < mhm; spent += notice {
|
||||
f.Speaking([]string{"ben", "ana"}) // the room ranks the steady mutter first
|
||||
talk(f, c, notice, both...)
|
||||
if got := f.Holder(); got != "ana" {
|
||||
t.Fatalf("floor went to %q after %v of backchannel; a rival owes %v",
|
||||
got, spent+notice, grab)
|
||||
}
|
||||
}
|
||||
talk(f, c, 100*time.Millisecond, both...)
|
||||
f.Speaking([]string{"ana"}) // ben stops; ana never did
|
||||
|
||||
talk(f, c, 200*time.Millisecond, voice{"ana", 1})
|
||||
@@ -255,6 +257,32 @@ func TestTurnBeginsBeforeWeKnewWhoseItWas(t *testing.T) {
|
||||
float64(said)/rate*1000)
|
||||
}
|
||||
|
||||
// Talk-over that its speaker then wins the floor with is not lost — it is handed
|
||||
// to their turn out of `lead`. A count that still calls it a loss overstates the
|
||||
// only number anyone has for how much this design throws away.
|
||||
func TestTalkOverThatWinsTheFloorIsNotALoss(t *testing.T) {
|
||||
c, s := started(), &scrap{}
|
||||
f := chair(c, s)
|
||||
|
||||
talk(f, c, 200*time.Millisecond, voice{"ana", 1})
|
||||
f.Speaking([]string{"ana"})
|
||||
|
||||
both := []voice{{"ana", 1}, {"ben", 2}}
|
||||
talk(f, c, notice, both...) // ben starts; the room has not ranked him yet
|
||||
f.Speaking([]string{"ben", "ana"})
|
||||
said := talk(f, c, grab, both...) / 2
|
||||
f.Speaking([]string{"ben", "ana"}) // and now he takes it
|
||||
|
||||
if got := f.Tally().Over; got != 0 {
|
||||
t.Fatalf("%.0fms called talk-over after all %.0fms of it was transcribed",
|
||||
float64(got)/rate*1000, float64(said)/rate*1000)
|
||||
}
|
||||
turns := s.got()
|
||||
if len(turns) != 2 || turns[1].marks()[2] < said {
|
||||
t.Fatalf("ben's turn holds %d of the %d he spoke over ana", turns[1].marks()[2], said)
|
||||
}
|
||||
}
|
||||
|
||||
// A real room keeps naming a speaker after they have stopped. Measured against
|
||||
// the live SFU: ana was still ranked first thirteen seconds after her last
|
||||
// packet, while ben was talking. A floor that trusts the ranking alone gives
|
||||
@@ -460,3 +488,71 @@ func TestAListenerNeedNotWatchTheFloor(t *testing.T) {
|
||||
a := &Attendant{heard: &collector{}}
|
||||
a.speaking([]lksdk.Participant{&lksdk.LocalParticipant{}}) // must not panic
|
||||
}
|
||||
|
||||
// A speaker who loses the floor and takes it back inside `lead` still has those
|
||||
// seconds in hand. Handed over a second time, the notes say what they said
|
||||
// twice — and the audio is billed twice for the privilege.
|
||||
func TestAudioIsGivenToOneTurnOnly(t *testing.T) {
|
||||
c, s := started(), &scrap{}
|
||||
f := chair(c, s)
|
||||
|
||||
// ana holds and talks.
|
||||
said := talk(f, c, 400*time.Millisecond, voice{"ana", 1})
|
||||
f.Speaking([]string{"ana"})
|
||||
said += talk(f, c, 400*time.Millisecond, voice{"ana", 1})
|
||||
|
||||
// She stops; the floor frees, and she starts again well inside `lead`.
|
||||
c.pass(rest + beat)
|
||||
f.look()
|
||||
if f.Holder() != "" {
|
||||
t.Fatal("floor still held after rest")
|
||||
}
|
||||
said += talk(f, c, 200*time.Millisecond, voice{"ana", 1})
|
||||
f.Speaking([]string{"ana"})
|
||||
|
||||
if err := f.Close(); err != nil {
|
||||
t.Fatalf("close: %v", err)
|
||||
}
|
||||
got := 0
|
||||
for _, turn := range s.got() {
|
||||
got += turn.marks()[1]
|
||||
}
|
||||
if got != said {
|
||||
t.Fatalf("%d samples reached a turn out of %d she spoke; %.0fms was transcribed twice",
|
||||
got, said, float64(got-said)/rate*1000)
|
||||
}
|
||||
if over := f.Tally().Over; over < 0 {
|
||||
t.Fatalf("talk-over counted %d, below zero: audio was given back more than once", over)
|
||||
}
|
||||
}
|
||||
|
||||
// A rival is not charged for our latency twice. They start talking; the room
|
||||
// takes an interval to rank them, and may take longer if the person they are
|
||||
// interrupting trails off slowly. The wait runs from when they STARTED, so what
|
||||
// they said in the meantime is inside `lead` and their turn opens on their first
|
||||
// word rather than a second into it.
|
||||
func TestTheWaitRunsFromWhenTheyStartedTalking(t *testing.T) {
|
||||
c, s := started(), &scrap{}
|
||||
f := chair(c, s)
|
||||
|
||||
talk(f, c, 200*time.Millisecond, voice{"ana", 1})
|
||||
f.Speaking([]string{"ana"})
|
||||
|
||||
// ben starts, and the room goes on ranking ana first for a full second
|
||||
// after — which is what a speaker trailing off looks like.
|
||||
both := []voice{{"ana", 1}, {"ben", 2}}
|
||||
said := talk(f, c, time.Second, both...) / 2
|
||||
f.Speaking([]string{"ben", "ana"})
|
||||
said += talk(f, c, grab-time.Second, both...) / 2
|
||||
f.Speaking([]string{"ben", "ana"})
|
||||
|
||||
turns := s.got()
|
||||
if len(turns) != 2 || turns[1].speaker != "ben" {
|
||||
t.Fatalf("turns %v after %v of ben talking; the wait is %v from when he started",
|
||||
turns, grab, grab)
|
||||
}
|
||||
if got := turns[1].marks()[2]; got != said {
|
||||
t.Fatalf("ben's turn opens %.0fms in: %d of the %d samples he had spoken",
|
||||
float64(said-got)/rate*1000, got, said)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -36,7 +36,7 @@ try() { # name, test, file, from, to
|
||||
}
|
||||
|
||||
try 'turn opens mid-word' TestTurnBeginsBeforeWeKnewWhoseItWas floor.go \
|
||||
'lead = 1200 * time.Millisecond' 'lead = 200 * time.Millisecond'
|
||||
'lead = 2500 * time.Millisecond' 'lead = 200 * time.Millisecond'
|
||||
|
||||
try 'no hysteresis' TestBackchannelDoesNotTakeTheFloor floor.go \
|
||||
'return top, now.Sub(f.from) >= grab' 'return top, true'
|
||||
@@ -69,4 +69,72 @@ try 'turn left open at the end' TestCancellingSettlesTheOpenTurn floor.go \
|
||||
f.Close()' ' case <-ctx.Done():
|
||||
_ = f'
|
||||
|
||||
|
||||
|
||||
try 'chunks the wrong size' TestAPushIsWholeFramesInsideTheCeiling scribe.go \
|
||||
'for len(t.pending) >= chunk {' 'for len(t.pending) >= chunk*3 {'
|
||||
|
||||
try 'no rollover at the limit' TestALongTurnRollsOverRatherThanBeingRefused scribe.go \
|
||||
'if t.sent+float64(len(pcm))/(rate*2) > span.Seconds() {' 'if false {'
|
||||
|
||||
try 'notes read out of order' TestNotesReadInTheOrderThingsWereSaid scribe.go \
|
||||
'seq := n.next
|
||||
n.next++' 'seq := -n.next
|
||||
n.next++'
|
||||
|
||||
try 'dropped audio hidden' TestAServiceThatCannotKeepUpLosesAudioOutLoud scribe.go \
|
||||
' default:
|
||||
t.to.mu.Lock()
|
||||
t.to.lost++
|
||||
t.to.mu.Unlock()' ' default:
|
||||
t.to.mu.Lock()
|
||||
t.to.mu.Unlock()'
|
||||
|
||||
# A select with a default never waits, so adding a timeout case to it changes
|
||||
# nothing — the first version of this mutation was a no-op and the run said so.
|
||||
# Blocking Hear means taking the default away.
|
||||
try 'Hear waits for the service' TestHearDoesNotWaitForTheService scribe.go \
|
||||
' select {
|
||||
case t.in <- append([]int16(nil), pcm...):
|
||||
default:
|
||||
t.to.mu.Lock()
|
||||
t.to.lost++
|
||||
t.to.mu.Unlock()
|
||||
}' ' t.in <- append([]int16(nil), pcm...)'
|
||||
|
||||
try 'recovered audio still called lost' TestTalkOverThatWinsTheFloorIsNotALoss floor.go \
|
||||
' if held[i].over {
|
||||
f.tally.Over -= len(held[i].pcm) // spoken over the last holder, kept anyway
|
||||
}' ' if false {
|
||||
f.tally.Over -= len(held[i].pcm)
|
||||
}'
|
||||
|
||||
|
||||
# The measured failure, pinned to the constant that fixes it: a rival that must
|
||||
# wait only one observer interval loses to a "mhm, yeah".
|
||||
try 'grab is one interval' TestBackchannelDoesNotTakeTheFloor floor.go \
|
||||
'grab = 2 * time.Second' 'grab = 500 * time.Millisecond'
|
||||
|
||||
|
||||
try 'opening burst frame by frame' TestTheOpeningBurstDoesNotOverrunTheTurn floor.go \
|
||||
' turn.Hear(began)' ' for i := 0; i < len(began); i += 320 {
|
||||
stop := i + 320
|
||||
if stop > len(began) {
|
||||
stop = len(began)
|
||||
}
|
||||
turn.Hear(began[i:stop])
|
||||
}'
|
||||
|
||||
|
||||
try 'audio given to two turns' TestAudioIsGivenToOneTurnOnly floor.go \
|
||||
' if held[i].sent {
|
||||
continue
|
||||
}' ' if false {
|
||||
continue
|
||||
}'
|
||||
|
||||
|
||||
try 'wait starts at the report' TestTheWaitRunsFromWhenTheyStartedTalking floor.go \
|
||||
'f.rival, f.from = top, f.began(top, now)' 'f.rival, f.from = top, now'
|
||||
|
||||
echo 'every test earns its keep'
|
||||
|
||||
+349
@@ -0,0 +1,349 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// service is hanzoai/speech's transcript API, answering the way it answers: a
|
||||
// session per open, a limit past which a push is refused, and the text on close.
|
||||
type service struct {
|
||||
mu sync.Mutex
|
||||
next int
|
||||
got map[string][]int // bytes of each push, per open session
|
||||
kept map[string][]int // and per session already closed
|
||||
refused int // pushes answered 409
|
||||
hold chan struct{} // when non-nil, pushes wait on it
|
||||
limit float64 // seconds a session accepts before 409, 0 for no limit
|
||||
}
|
||||
|
||||
func serve(t *testing.T, limit float64) (*service, string) {
|
||||
t.Helper()
|
||||
s := &service{got: map[string][]int{}, kept: map[string][]int{}, limit: limit}
|
||||
srv := httptest.NewServer(http.HandlerFunc(s.handle))
|
||||
t.Cleanup(srv.Close)
|
||||
return s, srv.URL
|
||||
}
|
||||
|
||||
func (s *service) handle(w http.ResponseWriter, r *http.Request) {
|
||||
id := strings.TrimPrefix(r.URL.Path, "/v1/audio/transcript")
|
||||
id = strings.TrimPrefix(id, "/")
|
||||
|
||||
switch {
|
||||
case r.Method == http.MethodPost && id == "":
|
||||
var ask map[string]any
|
||||
if err := json.NewDecoder(r.Body).Decode(&ask); err != nil {
|
||||
http.Error(w, err.Error(), 400)
|
||||
return
|
||||
}
|
||||
if ask["format"] != "pcm16" || ask["rate"] != float64(rate) || ask["channels"] != float64(1) {
|
||||
http.Error(w, fmt.Sprintf("wrong shape: %v", ask), 400)
|
||||
return
|
||||
}
|
||||
s.mu.Lock()
|
||||
s.next++
|
||||
id = fmt.Sprintf("atr_%d", s.next)
|
||||
s.got[id] = nil
|
||||
s.mu.Unlock()
|
||||
w.WriteHeader(201)
|
||||
json.NewEncoder(w).Encode(state{ID: id})
|
||||
|
||||
case r.Method == http.MethodPost:
|
||||
body, _ := io.ReadAll(r.Body)
|
||||
s.mu.Lock()
|
||||
hold, seen := s.hold, s.got[id]
|
||||
if seen == nil {
|
||||
if _, live := s.got[id]; !live {
|
||||
s.mu.Unlock()
|
||||
http.Error(w, "no transcript "+id, 404)
|
||||
return
|
||||
}
|
||||
}
|
||||
var sent int
|
||||
for _, n := range seen {
|
||||
sent += n
|
||||
}
|
||||
s.mu.Unlock()
|
||||
if hold != nil {
|
||||
<-hold
|
||||
}
|
||||
if s.limit > 0 && float64(sent+len(body))/(rate*2) > s.limit {
|
||||
s.mu.Lock()
|
||||
s.refused++
|
||||
s.mu.Unlock()
|
||||
http.Error(w, "at its limit", 409)
|
||||
return
|
||||
}
|
||||
s.mu.Lock()
|
||||
s.got[id] = append(s.got[id], len(body))
|
||||
s.mu.Unlock()
|
||||
json.NewEncoder(w).Encode(state{ID: id})
|
||||
|
||||
case r.Method == http.MethodDelete:
|
||||
s.mu.Lock()
|
||||
pushes, live := s.got[id]
|
||||
if !live {
|
||||
s.mu.Unlock()
|
||||
http.Error(w, "no transcript "+id, 404)
|
||||
return
|
||||
}
|
||||
var sent int
|
||||
for _, n := range pushes {
|
||||
sent += n
|
||||
}
|
||||
s.kept[id] = pushes
|
||||
delete(s.got, id)
|
||||
s.mu.Unlock()
|
||||
json.NewEncoder(w).Encode(state{ID: id, Text: id, Seconds: float64(sent) / (rate * 2)})
|
||||
|
||||
default:
|
||||
http.Error(w, "no", 405)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *service) pushes(id string) []int {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if open, live := s.got[id]; live {
|
||||
return append([]int(nil), open...)
|
||||
}
|
||||
return append([]int(nil), s.kept[id]...)
|
||||
}
|
||||
|
||||
// taken is the audio the service has actually accepted, in seconds.
|
||||
func (s *service) taken() float64 {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
sent := 0
|
||||
for _, all := range []map[string][]int{s.got, s.kept} {
|
||||
for _, pushes := range all {
|
||||
for _, n := range pushes {
|
||||
sent += n
|
||||
}
|
||||
}
|
||||
}
|
||||
return float64(sent) / (rate * 2)
|
||||
}
|
||||
|
||||
// The service refuses a body larger than its ceiling and one that is not whole
|
||||
// int16 frames, rather than truncating either. A turn that pushes the wrong
|
||||
// shape gets a 400 for every chunk and transcribes nothing.
|
||||
func TestAPushIsWholeFramesInsideTheCeiling(t *testing.T) {
|
||||
s, url := serve(t, 0)
|
||||
notes := Record(url, "whisper", "en")
|
||||
|
||||
turn, err := notes.Turn("ana", time.Now())
|
||||
if err != nil {
|
||||
t.Fatalf("open: %v", err)
|
||||
}
|
||||
// Two and a half chunks, so both the full pushes and the remainder are seen.
|
||||
for i := 0; i < 5; i++ {
|
||||
turn.Hear(make([]int16, chunk/2/2))
|
||||
}
|
||||
if err := turn.Close(); err != nil {
|
||||
t.Fatalf("close: %v", err)
|
||||
}
|
||||
|
||||
sent := s.pushes("atr_1")
|
||||
if len(sent) != 3 {
|
||||
t.Fatalf("%d pushes for 2.5 chunks of audio: %v", len(sent), sent)
|
||||
}
|
||||
for i, n := range sent {
|
||||
if n%2 != 0 {
|
||||
t.Fatalf("push %d is %d bytes, not whole int16 frames", i, n)
|
||||
}
|
||||
if n > chunk {
|
||||
t.Fatalf("push %d is %d bytes, over the %d ceiling", i, n, chunk)
|
||||
}
|
||||
}
|
||||
if sent[0] != chunk || sent[1] != chunk || sent[2] != chunk/2 {
|
||||
t.Fatalf("pushes %v; wanted two full chunks and the remainder", sent)
|
||||
}
|
||||
}
|
||||
|
||||
// A session takes 600 s of audio and then answers 409. A keynote is longer than
|
||||
// that, so the turn rolls into a fresh session and keeps what the last one
|
||||
// settled — otherwise the second half of a long turn is transcribed by nobody
|
||||
// and every push after the limit is an error nobody reads.
|
||||
func TestALongTurnRollsOverRatherThanBeingRefused(t *testing.T) {
|
||||
s, url := serve(t, span.Seconds())
|
||||
notes := Record(url, "whisper", "en")
|
||||
|
||||
turn, err := notes.Turn("ana", time.Now())
|
||||
if err != nil {
|
||||
t.Fatalf("open: %v", err)
|
||||
}
|
||||
// Ten seconds past the limit, a second at a time, never more than a minute
|
||||
// ahead of the service — which is what a room does, since audio arrives on a
|
||||
// clock and cannot arrive faster than it is spoken.
|
||||
const over = 10
|
||||
for fed := 0.0; fed < span.Seconds()+over; fed++ {
|
||||
turn.Hear(make([]int16, rate))
|
||||
for s.taken() < fed-60 {
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
}
|
||||
if err := turn.Close(); err != nil {
|
||||
t.Fatalf("close: %v", err)
|
||||
}
|
||||
|
||||
said := notes.Said()
|
||||
if len(said) != 1 {
|
||||
t.Fatalf("%d turns; a rollover is one turn, not two", len(said))
|
||||
}
|
||||
if s.next < 2 {
|
||||
t.Fatalf("%d sessions opened; %v of audio does not fit in one", s.next, span)
|
||||
}
|
||||
// Every second submitted is accounted for, across both sessions.
|
||||
if s.refused > 0 {
|
||||
t.Fatalf("%d pushes answered 409; the turn rolled over too late", s.refused)
|
||||
}
|
||||
if notes.Lost() > 0 {
|
||||
t.Fatalf("%d frames dropped feeding the turn", notes.Lost())
|
||||
}
|
||||
if got := said[0].Seconds; got < span.Seconds()+over-1 {
|
||||
t.Fatalf("turn reports %.1fs of %.1fs submitted; the rest went nowhere",
|
||||
got, span.Seconds()+over)
|
||||
}
|
||||
// And the text of the session that was closed on the way is still there.
|
||||
if !strings.Contains(said[0].Text, "atr_1") || !strings.Contains(said[0].Text, "atr_2") {
|
||||
t.Fatalf("text %q lost a session", said[0].Text)
|
||||
}
|
||||
}
|
||||
|
||||
// A turn that runs long settles after the short one that followed it. Notes read
|
||||
// in the order things were SAID, so they are ordered by when a turn began.
|
||||
func TestNotesReadInTheOrderThingsWereSaid(t *testing.T) {
|
||||
_, url := serve(t, 0)
|
||||
notes := Record(url, "whisper", "en")
|
||||
|
||||
began := time.Now()
|
||||
first, err := notes.Turn("ana", began)
|
||||
if err != nil {
|
||||
t.Fatalf("open: %v", err)
|
||||
}
|
||||
second, err := notes.Turn("ben", began.Add(time.Second))
|
||||
if err != nil {
|
||||
t.Fatalf("open: %v", err)
|
||||
}
|
||||
first.Hear(make([]int16, rate))
|
||||
second.Hear(make([]int16, rate))
|
||||
|
||||
if err := second.Close(); err != nil { // ben settles first
|
||||
t.Fatalf("close ben: %v", err)
|
||||
}
|
||||
if err := first.Close(); err != nil {
|
||||
t.Fatalf("close ana: %v", err)
|
||||
}
|
||||
|
||||
said := notes.Said()
|
||||
if len(said) != 2 || said[0].Speaker != "ana" || said[1].Speaker != "ben" {
|
||||
t.Fatalf("notes read %v; ana spoke first", said)
|
||||
}
|
||||
}
|
||||
|
||||
// A service too slow to keep up must slow the record, never the room. Frames
|
||||
// that cannot be queued are dropped and COUNTED: audio discarded quietly is a
|
||||
// transcript that reads complete and is not.
|
||||
func TestAServiceThatCannotKeepUpLosesAudioOutLoud(t *testing.T) {
|
||||
s, url := serve(t, 0)
|
||||
s.hold = make(chan struct{})
|
||||
notes := Record(url, "whisper", "en")
|
||||
|
||||
turn, err := notes.Turn("ana", time.Now())
|
||||
if err != nil {
|
||||
t.Fatalf("open: %v", err)
|
||||
}
|
||||
for i := 0; i < queue*3; i++ {
|
||||
turn.Hear(make([]int16, chunk)) // one chunk of frames each, well past the queue
|
||||
}
|
||||
lost := notes.Lost()
|
||||
|
||||
// Let the service go before asserting anything. A push still waiting is a
|
||||
// request the test server will not shut down without, so an assertion that
|
||||
// failed here would hang the run instead of reporting itself.
|
||||
close(s.hold)
|
||||
if err := turn.Close(); err != nil {
|
||||
t.Fatalf("close: %v", err)
|
||||
}
|
||||
if lost == 0 {
|
||||
t.Fatal("a stalled service dropped no frames and reported none")
|
||||
}
|
||||
t.Logf("%d frames dropped while the service was held", lost)
|
||||
}
|
||||
|
||||
// Hear must not block on the network. Every speaker's track is behind that call,
|
||||
// so a turn that waits for a push stalls the room, and audio arriving on a
|
||||
// real-time clock while nothing reads it is audio gone.
|
||||
func TestHearDoesNotWaitForTheService(t *testing.T) {
|
||||
s, url := serve(t, 0)
|
||||
s.hold = make(chan struct{})
|
||||
defer close(s.hold)
|
||||
notes := Record(url, "whisper", "en")
|
||||
|
||||
turn, err := notes.Turn("ana", time.Now())
|
||||
if err != nil {
|
||||
t.Fatalf("open: %v", err)
|
||||
}
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
for i := 0; i < queue*3; i++ {
|
||||
turn.Hear(make([]int16, chunk))
|
||||
}
|
||||
}()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("Hear blocked on a service that is not answering")
|
||||
}
|
||||
}
|
||||
|
||||
// A turn opens with everything its speaker said before the floor was theirs —
|
||||
// `lead` seconds arriving at once, which is not a rate any room delivers at. A
|
||||
// queue sized for a room drops what it cannot hold, and what it cannot hold is
|
||||
// the front of the turn: measured live, cleo's notes began "owe you the
|
||||
// capacity numbers".
|
||||
func TestTheOpeningBurstDoesNotOverrunTheTurn(t *testing.T) {
|
||||
_, url := serve(t, 0)
|
||||
notes := Record(url, "whisper", "en")
|
||||
c := started()
|
||||
f := chair(c, notes)
|
||||
|
||||
talk(f, c, 200*time.Millisecond, voice{"ana", 1})
|
||||
f.Speaking([]string{"ana"})
|
||||
|
||||
// ben talks under ana for the whole of the delay plus the wait, so his turn
|
||||
// opens holding every one of those seconds.
|
||||
both := []voice{{"ana", 1}, {"ben", 2}}
|
||||
talk(f, c, notice, both...)
|
||||
f.Speaking([]string{"ben", "ana"})
|
||||
said := talk(f, c, grab, both...) / 2
|
||||
|
||||
// Measured across the handover alone. This replay runs the meeting far
|
||||
// faster than a room does, so ordinary frames outrun the pusher here in a
|
||||
// way they cannot in a room; the burst is what a room DOES do, and it is
|
||||
// what this is about.
|
||||
before := notes.Lost()
|
||||
f.Speaking([]string{"ben", "ana"})
|
||||
if got := notes.Lost() - before; got != 0 {
|
||||
t.Fatalf("%d frames dropped opening a turn on %.1fs of held audio; "+
|
||||
"the first thing lost is the first thing said", got, float64(said)/rate)
|
||||
}
|
||||
if err := f.Close(); err != nil {
|
||||
t.Fatalf("close: %v", err)
|
||||
}
|
||||
for _, s := range notes.Said() {
|
||||
if s.Speaker == "ben" && s.Seconds < float64(said)/rate {
|
||||
t.Fatalf("ben's turn received %.2fs of the %.2fs he had already spoken",
|
||||
s.Seconds, float64(said)/rate)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user