nandi/frq · 9fcd0b4

nandi/frqpublic⑂ Fork 0

⑂ 9fcd0b4

Say when the call went away, and why

The last thing frq.av reads out of the media plane. joltmoq answered a code
per poll_status call and a reason from status_text, looped until it said
none; poll-status! answers the same thing as a drained queue of events, with
the reason and the camera/mic flags in the value instead of in three more
calls that had to be made in the right order.

Drained, not sampled. A call can fail and end between two pumps, and a
caller shown only the latest state would show the wrong one -- which is why
frq.av loops on it and why this queues rather than holding a current value.

A session that goes away says so through a FUTURE, not a callback: closed()
settles quietly on a clean close and raises the MoqError on a dirty one, so
completing it is how the reason is learned. Without it the only signal is
frames stopping, which can only ever say "something".

pump! is wrapped now. glimmer calls it from the loop thread, and a raise
escaping would take the window with it -- so a failure in the media plane
becomes a :failed event and the plane stays UP afterwards. joltmoq did the
same and the reason is the person: tearing down here would take the message
saying what went wrong off the screen along with it. frq.av decides when to
stop.

Writing the test found what a lost call actually looks like. Cancelling the
relay only stops it ACCEPTING; an established QUIC session then sits there
until its idle timeout, which is half a minute of frozen picture and no
event. What a peer hanging up looks like is that peer's own session being
cancelled, and the client sees it as `transport: webtransport error:
closed: code=0 reason=remote error: code=0` -- a transport error rather
than a graceful end, even though the far side passed code 0. Both :ended
and :failed are accepted as an ending; silence is not.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

nandi committed 2026-09-10T04:54:02-07:00 Browse files

9fcd0b4 parent: 596dc29

modified src/frq/av/plane.clj +68 -6 Viewed

@@ -11,6 +11,13 @@
11	11	start! stop! live? joltmoq_start / _stop / _is_live
12	12	poll-status! joltmoq_poll_status + _status_text
13	13	poll-frames! joltmoq_frame_poll + _frame_rgba
	14	+ poll-audio! (no equivalent; joltmoq played it itself)
	15	+
	16	+ STATUS IS A QUEUE, drained rather than sampled. joltmoq's poll_status
	17	+ answered one code per call and was looped until it said none, and the
	18	+ reason is worth keeping: a call can fail and end between two pumps, and a
	19	+ caller that only ever sees the latest state would show the wrong one.
	20	+ `frq.av` already loops on it.
14	21
15	22	WHY IT IS PUMPED AND NOT THREADED. jolt has fibers, but a fiber is bound to
16	23	its carrier for life and a blocking foreign call pins that carrier and
@@ -117,6 +124,12 @@
117	124	:track track
118	125	:path path
119	126	:session session
	127	+ ;; A session that closes says so through a future rather than
	128	+ ;; a callback. Completing it is how the REASON for a dropped
	129	+ ;; call is learned — it settles quietly on a clean close and
	130	+ ;; raises the MoqError on a dirty one — where noticing that
	131	+ ;; frames stopped would only ever say "something".
	132	+ :closed (when session (client/watch-closed! session))
120	133	;; The announcement watch is the whole of peer discovery. An
121	134	;; empty prefix takes everything on the origin, because in a
122	135	;; call every participant is a broadcast and none of their
@@ -147,7 +160,9 @@
147	160	:size [width height]
148	161	:camera? camera?
149	162	:frames []
150		- :status (atom [])
	163	+ :status (atom [{:code :live
	164	+ :has-camera? (some? source)
	165	+ :has-mic? (some? mic)}])
151	166	:pts (atom 0)
152	167	:fps fps})
153	168	true))
@@ -415,13 +430,44 @@
415	430
416	431	;; --- the pump ----------------------------------------------------------------
417	432
	433	+(defn- note!
	434	+ "Queue a status transition for the next `poll-status!`."
	435	+ [p event]
	436	+ (when-let [q (:status p)] (swap! q conj event))
	437	+ nil)
	438	+
	439	+(defn- pump-closed!
	440	+ "Notice the session going away, and why.
	441	+
	442	+ A clean shutdown settles the future with nothing; a dropped connection
	443	+ raises the MoqError that caused it. Both mean the call is over, and both
	444	+ have to be announced — the difference is only what the person is told."
	445	+ [p]
	446	+ (if-let [w (:closed p)]
	447	+ (if (uniffi/settled? w)
	448	+ (do (try
	449	+ (uniffi/complete! w)
	450	+ (note! p {:code :ended})
	451	+ (catch Exception e
	452	+ (note! p {:code :failed
	453	+ :text (or (:message (ex-data e)) (ex-message e))})))
	454	+ (assoc p :closed nil))
	455	+ p)
	456	+ p))
	457	+
418	458	(defn pump!
419		- "Drive both halves once. Called from the same timer as `frq.av/pump!`."
	459	+ "Drive both halves once. Called from the same timer as `frq.av/pump!`.
	460	+
	461	+ Everything inside is wrapped: a raise from the media plane is a call that
	462	+ failed, not a UI that stops repainting. glimmer calls this from the loop
	463	+ thread, and an exception escaping here would take the window with it."
420	464	[]
421	465	(when-let [p @plane]
422		- (pump-out! p)
423		- (pump-mic! p)
424		- (let [p' (pump-in! p)
	466	+ (try
	467	+ (let [p (pump-closed! p)]
	468	+ (pump-out! p)
	469	+ (pump-mic! p)
	470	+ (let [p' (pump-in! p)
425	471	;; Mix everyone EXCEPT ourselves: hearing your own voice back is
426	472	;; the thing headphones exist to prevent.
427	473	rings (keep (fn [[_ peer]] (when-not (:self? peer) (:ring peer)))
@@ -437,6 +483,14 @@
437	483	(when (and mixed (:speaker p'))
438	484	((:play! (:speaker p')) mixed))
439	485	(reset! plane (assoc p' :mixed mixed)))))
	486	+ (catch Exception e
	487	+ (note! p {:code :failed
	488	+ :text (or (:message (ex-data e)) (ex-message e))})
	489	+ ;; The plane stays UP after a failure. joltmoq did the same, and the
	490	+ ;; reason is the person: tearing it down here would take the message
	491	+ ;; saying what went wrong off the screen along with it. `frq.av`
	492	+ ;; decides when to stop.
	493	+ (reset! plane (assoc p :frames [] :mixed nil)))))
440	494	nil)
441	495
442	496	(defn poll-frames!
@@ -468,7 +522,15 @@
468	522	(some-> @plane :peers keys vec))
469	523
470	524	(defn poll-status!
471		- "Drain what the plane has learned, as [code text] pairs, oldest first."
	525	+ "Drain what the plane has learned since the last call, oldest first.
	526	+
	527	+ Each event is {:code :live\|:ended\|:failed} with `:text` on a failure and
	528	+ `:has-camera?`/`:has-mic?` on :live — which is joltmoq's poll_status,
	529	+ status_text and status_has_camera/_mic in one value instead of four
	530	+ calls that had to be made in the right order.
	531	+
	532	+ Drained, not sampled: a call can fail and end between two pumps, and a
	533	+ caller shown only the latest would show the wrong one."
472	534	[]
473	535	(when-let [p @plane]
474	536	(let [q (:status p)

        start! stop! live?          joltmoq_start / _stop / _is_live
        poll-status!                joltmoq_poll_status + _status_text
        poll-frames!                joltmoq_frame_poll + _frame_rgba
++      poll-audio!                 (no equivalent; joltmoq played it itself)
++
++  STATUS IS A QUEUE, drained rather than sampled. joltmoq's poll_status
++  answered one code per call and was looped until it said none, and the
++  reason is worth keeping: a call can fail and end between two pumps, and a
++  caller that only ever sees the latest state would show the wrong one.
++  `frq.av` already loops on it.
    WHY IT IS PUMPED AND NOT THREADED. jolt has fibers, but a fiber is bound to
    its carrier for life and a blocking foreign call pins that carrier and
               :track     track
               :path      path
               :session   session
++             ;; A session that closes says so through a future rather than
++             ;; a callback. Completing it is how the REASON for a dropped
++             ;; call is learned — it settles quietly on a clean close and
++             ;; raises the MoqError on a dirty one — where noticing that
++             ;; frames stopped would only ever say "something".
++             :closed    (when session (client/watch-closed! session))
               ;; The announcement watch is the whole of peer discovery. An
               ;; empty prefix takes everything on the origin, because in a
               ;; call every participant is a broadcast and none of their
               :size      [width height]
               :camera?   camera?
               :frames    []
--             :status    (atom [])
++             :status    (atom [{:code :live
++                                :has-camera? (some? source)
++                                :has-mic?    (some? mic)}])
               :pts       (atom 0)
               :fps       fps})
      true))
  ;; --- the pump ----------------------------------------------------------------
++(defn- note!
++  "Queue a status transition for the next `poll-status!`."
++  [p event]
++  (when-let [q (:status p)] (swap! q conj event))
++  nil)
++
++(defn- pump-closed!
++  "Notice the session going away, and why.
++
++  A clean shutdown settles the future with nothing; a dropped connection
++  raises the MoqError that caused it. Both mean the call is over, and both
++  have to be announced — the difference is only what the person is told."
++  [p]
++  (if-let [w (:closed p)]
++    (if (uniffi/settled? w)
++      (do (try
++            (uniffi/complete! w)
++            (note! p {:code :ended})
++            (catch Exception e
++              (note! p {:code :failed
++                        :text (or (:message (ex-data e)) (ex-message e))})))
++          (assoc p :closed nil))
++      p)
++    p))
++
  (defn pump!
--  "Drive both halves once. Called from the same timer as `frq.av/pump!`."
++  "Drive both halves once. Called from the same timer as `frq.av/pump!`.
++
++  Everything inside is wrapped: a raise from the media plane is a call that
++  failed, not a UI that stops repainting. glimmer calls this from the loop
++  thread, and an exception escaping here would take the window with it."
    []
    (when-let [p @plane]
--    (pump-out! p)
++   (try
--    (pump-mic! p)
++    (let [p (pump-closed! p)]
--    (let [p' (pump-in! p)
++     (pump-out! p)
++     (pump-mic! p)
++     (let [p' (pump-in! p)
            ;; Mix everyone EXCEPT ourselves: hearing your own voice back is
            ;; the thing headphones exist to prevent.
            rings (keep (fn [[_ peer]] (when-not (:self? peer) (:ring peer)))
          (when (and mixed (:speaker p'))
            ((:play! (:speaker p')) mixed))
          (reset! plane (assoc p' :mixed mixed)))))
++    (catch Exception e
++      (note! p {:code :failed
++                :text (or (:message (ex-data e)) (ex-message e))})
++      ;; The plane stays UP after a failure. joltmoq did the same, and the
++      ;; reason is the person: tearing it down here would take the message
++      ;; saying what went wrong off the screen along with it. `frq.av`
++      ;; decides when to stop.
++      (reset! plane (assoc p :frames [] :mixed nil)))))
    nil)
  (defn poll-frames!
    (some-> @plane :peers keys vec))
  (defn poll-status!
--  "Drain what the plane has learned, as [code text] pairs, oldest first."
++  "Drain what the plane has learned since the last call, oldest first.
++
++  Each event is {:code :live|:ended|:failed} with `:text` on a failure and
++  `:has-camera?`/`:has-mic?` on :live — which is joltmoq's poll_status,
++  status_text and status_has_camera/_mic in one value instead of four
++  calls that had to be made in the right order.
++
++  Drained, not sampled: a call can fail and end between two pumps, and a
++  caller shown only the latest would show the wrong one."
    []
    (when-let [p @plane]
      (let [q (:status p)

modified src/frq/moq/client.clj +10 -0 Viewed

@@ -206,6 +206,16 @@
206	206	(uniffi/with-out-status #(raw/method-moqserver-cancel (clone-server server) %))
207	207	nil)
208	208
	209	+(defn watch-closed!
	210	+ "A :void future that settles when the session closes.
	211	+
	212	+ Settles QUIETLY on a clean close and RAISES on a dirty one, with the
	213	+ MoqError saying why — so completing it is how the reason for a dropped
	214	+ call is learned, rather than by noticing that frames stopped."
	215	+ [session]
	216	+ (-> (raw/method-moqsession-closed (clone-session session))
	217	+ (uniffi/start-future :void)))
	218	+
209	219	(defn shutdown!
210	220	"Graceful shutdown — equivalent to `(cancel! session 0)`.
211	221

    (uniffi/with-out-status #(raw/method-moqserver-cancel (clone-server server) %))
    nil)
++(defn watch-closed!
++  "A :void future that settles when the session closes.
++
++  Settles QUIETLY on a clean close and RAISES on a dirty one, with the
++  MoqError saying why — so completing it is how the reason for a dropped
++  call is learned, rather than by noticing that frames stopped."
++  [session]
++  (-> (raw/method-moqsession-closed (clone-session session))
++      (uniffi/start-future :void)))
++
  (defn shutdown!
    "Graceful shutdown — equivalent to `(cancel! session 0)`.

modified src/frq/moq/smoke.clj +102 -1 Viewed

@@ -716,6 +716,106 @@
716	716	((:close! c)))))
717	717	true)
718	718
	719	+(defn- check-status
	720	+ "The three transitions frq.av reads: live, failed, ended.
	721	+
	722	+ :live is asserted on a plain local plane — it carries has-camera? and
	723	+ has-mic?, which is what the UI shows before a single frame arrives.
	724	+
	725	+ :ended and :failed need a session to lose, so the relay from the session
	726	+ check is stood up again and then CANCELLED underneath a running plane.
	727	+ That is the case that matters: not a call the person hung up, but one
	728	+ that went away, which is the whole reason poll-status! exists rather than
	729	+ frq.av inferring things from frames stopping.
	730	+
	731	+ Whether the drop reads as :ended or :failed depends on how the far side
	732	+ goes — a relay cancelled mid-session may close cleanly or not — so both
	733	+ are accepted here. What is NOT accepted is silence: a call that ends with
	734	+ no transition at all is one where the person is left looking at a frozen
	735	+ picture."
	736	+ []
	737	+ (ffi/with-arena [a]
	738	+ ;; 1. :live, with the flags.
	739	+ (let [origin (media/new-origin)]
	740	+ (plane/start! {:origin origin :path "/us"
	741	+ :source (fn [] nil) :mic (fn [] nil)
	742	+ :width 64 :height 64 :channels 1})
	743	+ (let [[ev] (plane/poll-status!)]
	744	+ (println " live event:" (pr-str ev))
	745	+ (when-not (= :live (:code ev))
	746	+ (throw (ex-info "no :live on start" {:event ev})))
	747	+ (when-not (and (:has-camera? ev) (:has-mic? ev))
	748	+ (throw (ex-info "flags do not reflect the sources given" {:event ev})))
	749	+ (when (seq (plane/poll-status!))
	750	+ (throw (ex-info "poll-status! did not drain" {}))))
	751	+ (plane/stop!))
	752	+
	753	+ ;; 2. a session lost underneath us.
	754	+ (let [[px _] (i420-halves a 64 64 0x40 0xC0)
	755	+ relay (media/new-origin)
	756	+ server (client/new-server)]
	757	+ (client/server-bind! server "127.0.0.1:0")
	758	+ (client/server-tls-generate! server ["localhost"])
	759	+ (client/server-origin! server relay)
	760	+ (let [addr (settle! (client/server-listen! server) "listen" 10000
	761	+ uniffi/lift-string)
	762	+ port (last (str/split addr #":"))
	763	+ fps (client/server-fingerprints server)
	764	+ c (client/new-client)]
	765	+ (client/set-tls-fingerprints! c fps)
	766	+ (let [connect (client/connect! c (str "https://localhost:" port "/room"))
	767	+ incoming (client/server-accept! server)
	768	+ deadline (+ (System/currentTimeMillis) 25000)]
	769	+ (loop [req nil accepted nil srv nil sess nil]
	770	+ (let [req (or req (settle! incoming "accept" 0 media/lift-optional-handle))
	771	+ accepted (or accepted (when req (client/accept-request! req)))
	772	+ ;; The relay's OWN side of the session, kept rather than
	773	+ ;; dropped: cancelling the server only stops it accepting
	774	+ ;; new connections, and an established QUIC session then
	775	+ ;; sits there until its idle timeout — half a minute of
	776	+ ;; a frozen picture. What a peer hanging up actually
	777	+ ;; looks like is this session being cancelled.
	778	+ srv (or srv (when accepted
	779	+ (settle! accepted "request accept" 0 nil)))
	780	+ sess (or sess (settle! connect "connect" 0 nil))]
	781	+ (cond
	782	+ (and sess srv)
	783	+ (do
	784	+ (plane/start! {:origin (client/session-publisher sess)
	785	+ :discover (client/session-consumer sess)
	786	+ :session sess
	787	+ :path "/us" :source (fn [] [px nil])
	788	+ :width 64 :height 64 :fps 30 :bitrate 200000})
	789	+ (plane/poll-status!) ; drain the :live
	790	+ (dotimes [_ 10] (plane/pump!) (Thread/sleep 10))
	791	+ (println " dropping the far side under a live plane")
	792	+ (client/cancel! srv 0)
	793	+ (client/server-cancel! server)
	794	+ (let [d2 (+ (System/currentTimeMillis) 20000)]
	795	+ (loop [pumps 0]
	796	+ (plane/pump!)
	797	+ (let [evs (plane/poll-status!)]
	798	+ (cond
	799	+ (seq evs)
	800	+ (do (println " after the drop:" (pr-str evs))
	801	+ (when-not (some #{:ended :failed} (map :code evs))
	802	+ (throw (ex-info "the drop produced no ending"
	803	+ {:events evs})))
	804	+ (plane/stop!)
	805	+ true)
	806	+
	807	+ (> (System/currentTimeMillis) d2)
	808	+ (do (plane/stop!)
	809	+ (throw (ex-info "the session went away silently"
	810	+ {:pumps pumps})))
	811	+
	812	+ :else (do (Thread/sleep 10) (recur (inc pumps))))))))
	813	+
	814	+ (> (System/currentTimeMillis) deadline)
	815	+ (throw (ex-info "the session never came up" {}))
	816	+
	817	+ :else (do (Thread/sleep 10) (recur req accepted srv sess))))))))))
	818	+
719	819	(defn -main [& _]
720	820	(println "libmoq_ffi smoke test")
721	821	(let [steps [["contract" check-contract]
@@ -731,7 +831,8 @@
731	831	["plane" check-plane]
732	832	["audio" check-audio]
733	833	["session" check-session]
734		- ["wired" check-wired-devices]]]
	834	+ ["wired" check-wired-devices]
	835	+ ["status" check-status]]]
735	836	(doseq [[name f] steps]
736	837	(println (str name ":"))
737	838	(f))

          ((:close! c)))))
    true)
++(defn- check-status
++  "The three transitions frq.av reads: live, failed, ended.
++
++  :live is asserted on a plain local plane — it carries has-camera? and
++  has-mic?, which is what the UI shows before a single frame arrives.
++
++  :ended and :failed need a session to lose, so the relay from the session
++  check is stood up again and then CANCELLED underneath a running plane.
++  That is the case that matters: not a call the person hung up, but one
++  that went away, which is the whole reason poll-status! exists rather than
++  frq.av inferring things from frames stopping.
++
++  Whether the drop reads as :ended or :failed depends on how the far side
++  goes — a relay cancelled mid-session may close cleanly or not — so both
++  are accepted here. What is NOT accepted is silence: a call that ends with
++  no transition at all is one where the person is left looking at a frozen
++  picture."
++  []
++  (ffi/with-arena [a]
++    ;; 1. :live, with the flags.
++    (let [origin (media/new-origin)]
++      (plane/start! {:origin origin :path "/us"
++                     :source (fn [] nil) :mic (fn [] nil)
++                     :width 64 :height 64 :channels 1})
++      (let [[ev] (plane/poll-status!)]
++        (println "  live event:" (pr-str ev))
++        (when-not (= :live (:code ev))
++          (throw (ex-info "no :live on start" {:event ev})))
++        (when-not (and (:has-camera? ev) (:has-mic? ev))
++          (throw (ex-info "flags do not reflect the sources given" {:event ev})))
++        (when (seq (plane/poll-status!))
++          (throw (ex-info "poll-status! did not drain" {}))))
++      (plane/stop!))
++
++    ;; 2. a session lost underneath us.
++    (let [[px _]  (i420-halves a 64 64 0x40 0xC0)
++          relay   (media/new-origin)
++          server  (client/new-server)]
++      (client/server-bind! server "127.0.0.1:0")
++      (client/server-tls-generate! server ["localhost"])
++      (client/server-origin! server relay)
++      (let [addr (settle! (client/server-listen! server) "listen" 10000
++                          uniffi/lift-string)
++            port (last (str/split addr #":"))
++            fps  (client/server-fingerprints server)
++            c    (client/new-client)]
++        (client/set-tls-fingerprints! c fps)
++        (let [connect  (client/connect! c (str "https://localhost:" port "/room"))
++              incoming (client/server-accept! server)
++              deadline (+ (System/currentTimeMillis) 25000)]
++          (loop [req nil accepted nil srv nil sess nil]
++            (let [req      (or req (settle! incoming "accept" 0 media/lift-optional-handle))
++                  accepted (or accepted (when req (client/accept-request! req)))
++                  ;; The relay's OWN side of the session, kept rather than
++                  ;; dropped: cancelling the server only stops it accepting
++                  ;; new connections, and an established QUIC session then
++                  ;; sits there until its idle timeout — half a minute of
++                  ;; a frozen picture. What a peer hanging up actually
++                  ;; looks like is this session being cancelled.
++                  srv      (or srv (when accepted
++                                     (settle! accepted "request accept" 0 nil)))
++                  sess     (or sess (settle! connect "connect" 0 nil))]
++              (cond
++                (and sess srv)
++                (do
++                  (plane/start! {:origin   (client/session-publisher sess)
++                                 :discover (client/session-consumer sess)
++                                 :session  sess
++                                 :path "/us" :source (fn [] [px nil])
++                                 :width 64 :height 64 :fps 30 :bitrate 200000})
++                  (plane/poll-status!)              ; drain the :live
++                  (dotimes [_ 10] (plane/pump!) (Thread/sleep 10))
++                  (println "  dropping the far side under a live plane")
++                  (client/cancel! srv 0)
++                  (client/server-cancel! server)
++                  (let [d2 (+ (System/currentTimeMillis) 20000)]
++                    (loop [pumps 0]
++                      (plane/pump!)
++                      (let [evs (plane/poll-status!)]
++                        (cond
++                          (seq evs)
++                          (do (println "  after the drop:" (pr-str evs))
++                              (when-not (some #{:ended :failed} (map :code evs))
++                                (throw (ex-info "the drop produced no ending"
++                                                {:events evs})))
++                              (plane/stop!)
++                              true)
++
++                          (> (System/currentTimeMillis) d2)
++                          (do (plane/stop!)
++                              (throw (ex-info "the session went away silently"
++                                              {:pumps pumps})))
++
++                          :else (do (Thread/sleep 10) (recur (inc pumps))))))))
++
++                (> (System/currentTimeMillis) deadline)
++                (throw (ex-info "the session never came up" {}))
++
++                :else (do (Thread/sleep 10) (recur req accepted srv sess))))))))))
++
  (defn -main [& _]
    (println "libmoq_ffi smoke test")
    (let [steps [["contract" check-contract]
                 ["plane"    check-plane]
                 ["audio"    check-audio]
                 ["session"  check-session]
--               ["wired"    check-wired-devices]]]
++               ["wired"    check-wired-devices]
++               ["status"   check-status]]]
      (doseq [[name f] steps]
        (println (str name ":"))
        (f))