Merge the capture and mavpoll sweep (#224)

V-581. Session.discard replaces the finish-unlink-remove block that reapLocked and Abort each spelled out. One abandon closure covers the two Start failure paths that close and remove the spool. The third closes without removing, and that was left exactly as it was rather than silently changing what it deletes. bytesPerSample and bytesPerSecond replace three copies of the byte-rate arithmetic. The two spellings differed in where the divide by 8 fell, which only matters for a sample width that is not a multiple of 8 bits. Every format in the package is 16-bit, so this is arithmetically identical. In mavpoll, four writers each built their own WriteFactReq literal and three repeated a redundant error check, where isNoFact is errors.Is and already covered the equality arm. unchanged and writeFact hold that now, so each writer keeps only its log line. run lost the zenmoney setup and the ticker, so it is flag parsing and wiring. Two stale comments corrected. The package doc said two sources, listed three, and polls four. Filed V-583: CLAUDE.md's daemon table calls mavpoll the Telegram long-poll reach. It is the env poller. Telegram is internal/delivery/telegramsink.
2026-08-06 01:55:18 +04:00
parent 91f6ea84a0 db8cbdc20a
commit 60540fa934
2 changed files with 144 additions and 84 deletions
@@ -6,10 +6,11 @@
 // restart-free, fail-independent — a crashing poller can't touch the store key
 // (it never had it), worst case a stale env fact until the next tick.
 //
-// Two sources, each its own provenance (the loop's rules trust source):
+// Four sources, each its own provenance (the loop's rules trust source):
 //   - netdata  → poll:netdata      resource alarms (disk/mem/cert/temp)
 //   - kuma     → poll:uptimekuma   service up/down (the source of truth for it)
 //   - zenmoney → poll:zenmoney     spending/income totals (Vikunja #125)
+//   - wireguard → infer:wg         latest handshake, the presence signal
 //
 // The zenmoney source is why the token lives HERE and not in core: the poller
 // already owns every other third-party credential, it holds no store key, and
@@ -77,26 +78,15 @@ func run(args []string) error {
 		return fmt.Errorf("nothing to poll: set -netdata, -kuma, -wg and/or -zenmoney-token-file")
 	}

-	// The token is read from a file, never taken as a flag value: an argv token
-	// is visible in `ps` to every user on the box and lands in the compose file
-	// and the shell history. Read once at start — a rotated token means a
-	// restart, which is cheaper than re-reading his credential every hour.
-	var zen *zenmoney.Client
-	if *zenTokenFile != "" {
-		raw, err := os.ReadFile(*zenTokenFile)
-		if err != nil {
-			return fmt.Errorf("read zenmoney token: %w", err)
-		}
-		zen, err = zenmoney.New(strings.TrimSpace(string(raw)), *zenURL, *timeout*3)
-		if err != nil {
-			return err
-		}
+	zen, err := newZenClient(*zenTokenFile, *zenURL, *timeout)
+	if err != nil {
+		return err
 	}

 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer stop()

-	core, err := ipc.DialWait(*socket, 60*time.Second)
+	core, err := ipc.DialWait(*socket, coreDialWait)
 	if err != nil {
 		return err
 	}
@@ -117,14 +107,46 @@ func run(args []string) error {
 	// The token is never logged, not even its length.
 	log.Printf("mavpoll: polling every %s (netdata=%q kuma=%q wg=%q zenmoney=%v every %s)",
 		*interval, *netdataURL, *kumaURL, *wgIface, zen != nil, *zenInterval)
+	p.loop(ctx, *interval)
+	return nil
+}
+
+// coreDialWait — how long to wait for core's socket at start. The poller and
+// core come up together under compose, so a cold start is a wait, not a failure.
+const coreDialWait = 60 * time.Second
+
+// zenTimeoutFactor — the zenmoney client gets a longer deadline than the other
+// sources. A diff call walks his whole transaction history, where netdata and
+// kuma answer from memory.
+const zenTimeoutFactor = 3
+
+// newZenClient builds the money client, or nil when no token file was given.
+//
+// The token is read from a file, never taken as a flag value: an argv token is
+// visible in `ps` to every user on the box and lands in the compose file and
+// the shell history. Read once at start — a rotated token means a restart,
+// which is cheaper than re-reading his credential every hour.
+func newZenClient(tokenFile, baseURL string, timeout time.Duration) (*zenmoney.Client, error) {
+	if tokenFile == "" {
+		return nil, nil
+	}
+	raw, err := os.ReadFile(tokenFile)
+	if err != nil {
+		return nil, fmt.Errorf("read zenmoney token: %w", err)
+	}
+	return zenmoney.New(strings.TrimSpace(string(raw)), baseURL, timeout*zenTimeoutFactor)
+}
+
+// loop polls until the context is cancelled.
+func (p *poller) loop(ctx context.Context, interval time.Duration) {
 	p.pollOnce(ctx) // fire immediately; don't idle a full interval on start
-	t := time.NewTicker(*interval)
+	t := time.NewTicker(interval)
 	defer t.Stop()
 	for {
 		select {
 		case <-ctx.Done():
 			log.Printf("mavpoll: bye")
-			return nil
+			return
 		case <-t.C:
 			p.pollOnce(ctx)
 		}
@@ -151,8 +173,8 @@ type poller struct {
 	zenLast  time.Time
 }

-// pollOnce — one sweep of both sources. A failure in one source logs and does
-// NOT abort the other: netdata being down shouldn't blind kuma and vice versa.
+// pollOnce — one sweep of every configured source. A failure in one logs and
+// does NOT abort the rest: netdata being down shouldn't blind kuma.
 func (p *poller) pollOnce(ctx context.Context) {
 	now := time.Now()
 	if p.netdataURL != "" {
@@ -244,6 +266,14 @@ func (p *poller) pollZenmoney(ctx context.Context, now time.Time) error {

 // ---- wireguard: latest handshake → presence signal -------------------------

+const (
+	// wgFactKey / wgSource — the presence signal, read by the decay in core.
+	// The source says infer because a handshake is evidence he is home, not a
+	// reading of where he is.
+	wgFactKey = "wg_handshake"
+	wgSource  = "infer:wg"
+)
+
 // pollWg reads `wg show <iface> latest-handshakes` and writes a wg_handshake
 // fact (source=infer:wg) stamped with the MOST RECENT peer handshake time — not
 // now(). Presence decays from the real handshake instant, so the fact's ts must
@@ -263,20 +293,19 @@ func (p *poller) pollWg(ctx context.Context) error {
 		return nil // no peer has ever handshaked → drop out of presence
 	}
 	hs := time.Unix(maxTs, 0)
-	prev, err := p.core.LatestFactBySource(ctx, "wg_handshake", "infer:wg")
+	prev, err := p.core.LatestFactBySource(ctx, wgFactKey, wgSource)
 	if err == nil && !hs.After(prev.Ts) {
 		return nil // not newer → no churn
 	}
-	if err != nil && err != ipc.ErrNoFact && !isNoFact(err) {
-		return fmt.Errorf("read wg_handshake: %w", err)
+	if err != nil && !isNoFact(err) {
+		return fmt.Errorf("read %s: %w", wgFactKey, err)
 	}
-	if _, err := p.core.WriteFact(ctx, ipc.WriteFactReq{
-		Ts: hs, Kind: "env", Key: "wg_handshake", Value: `"up"`,
-		Source: "infer:wg", Confidence: 1.0,
-	}); err != nil {
-		return fmt.Errorf("write wg_handshake: %w", err)
+	// The ts is the handshake instant, not now(): presence decays from when he
+	// was last seen.
+	if err := p.writeFact(ctx, wgFactKey, wgSource, `"up"`, hs); err != nil {
+		return err
 	}
-	log.Printf("mavpoll: wg_handshake @ %s (infer:wg)", hs.Format(time.RFC3339))
+	log.Printf("mavpoll: %s @ %s (%s)", wgFactKey, hs.Format(time.RFC3339), wgSource)
 	return nil
 }

@@ -431,29 +460,53 @@ func kumaState(v float64) string {

 // ---- helpers ---------------------------------------------------------------

-// writeIfChanged writes a `facts(kind=env)` row only when val differs from the
-// latest fact for (key, source). Values are stored JSON-encoded (the store's
-// convention: `"down"`, `"critical"`), matching how rules compare f.Value.
-func (p *poller) writeIfChanged(ctx context.Context, key, source, val string, now time.Time) error {
-	jv, _ := json.Marshal(val) // string never fails to marshal
+// factConfidence — every poll is a direct reading of another service, never an
+// inference, so the fact goes in at full confidence.
+const factConfidence = 1.0
+
+// unchanged reports whether the latest fact for (key, source) already holds
+// jsonVal. A missing fact is not an error here, it is the first write.
+func (p *poller) unchanged(ctx context.Context, key, source, jsonVal string) (bool, error) {
 	prev, err := p.core.LatestFactBySource(ctx, key, source)
 	switch {
-	case err == nil && prev.Value == string(jv):
-		return nil // unchanged → no churn
-	case err != nil && err != ipc.ErrNoFact && !isNoFact(err):
-		return fmt.Errorf("read %s: %w", key, err)
+	case err == nil:
+		return prev.Value == jsonVal, nil
+	case isNoFact(err):
+		return false, nil
+	default:
+		return false, fmt.Errorf("read %s: %w", key, err)
 	}
-	_, err = p.core.WriteFact(ctx, ipc.WriteFactReq{
+}
+
+// writeFact writes one `facts(kind=env)` row. Every poll in this file lands
+// here, so the row shape is written once.
+func (p *poller) writeFact(ctx context.Context, key, source, jsonVal string, now time.Time) error {
+	_, err := p.core.WriteFact(ctx, ipc.WriteFactReq{
 		Ts:         now,
 		Kind:       "env",
 		Key:        key,
-		Value:      string(jv),
+		Value:      jsonVal,
 		Source:     source,
-		Confidence: 1.0, // a direct reading, not an inference
+		Confidence: factConfidence,
 	})
 	if err != nil {
 		return fmt.Errorf("write %s: %w", key, err)
 	}
+	return nil
+}
+
+// writeIfChanged writes only when val differs from the latest fact for
+// (key, source). Values are stored JSON-encoded (the store's convention:
+// `"down"`, `"critical"`), matching how rules compare f.Value.
+func (p *poller) writeIfChanged(ctx context.Context, key, source, val string, now time.Time) error {
+	jv, _ := json.Marshal(val) // a string never fails to marshal
+	same, err := p.unchanged(ctx, key, source, string(jv))
+	if err != nil || same {
+		return err // unchanged → no churn
+	}
+	if err := p.writeFact(ctx, key, source, string(jv), now); err != nil {
+		return err
+	}
 	log.Printf("mavpoll: %s=%s (%s)", key, val, source)
 	return nil
 }
@@ -466,18 +519,12 @@ func (p *poller) writeIfChanged(ctx context.Context, key, source, val string, no
 // The log line names the key and the source, never the figures: mavpoll's log
 // is not the place his spending ends up.
 func (p *poller) writeIfChangedRaw(ctx context.Context, key, source, jsonVal string, now time.Time) error {
-	prev, err := p.core.LatestFactBySource(ctx, key, source)
-	switch {
-	case err == nil && prev.Value == jsonVal:
-		return nil
-	case err != nil && err != ipc.ErrNoFact && !isNoFact(err):
-		return fmt.Errorf("read %s: %w", key, err)
+	same, err := p.unchanged(ctx, key, source, jsonVal)
+	if err != nil || same {
+		return err
 	}
-	if _, err := p.core.WriteFact(ctx, ipc.WriteFactReq{
-		Ts: now, Kind: "env", Key: key, Value: jsonVal,
-		Source: source, Confidence: 1.0,
-	}); err != nil {
-		return fmt.Errorf("write %s: %w", key, err)
+	if err := p.writeFact(ctx, key, source, jsonVal, now); err != nil {
+		return err
 	}
 	log.Printf("mavpoll: %s updated (%s)", key, source)
 	return nil
@@ -490,11 +537,8 @@ func (p *poller) writeIfChangedRaw(ctx context.Context, key, source, jsonVal str
 // The log line names the key only, never the figures: mavpoll's log is not the
 // place his spending ends up.
 func (p *poller) writeMoneyFact(ctx context.Context, key, jsonVal string, now time.Time) error {
-	if _, err := p.core.WriteFact(ctx, ipc.WriteFactReq{
-		Ts: now, Kind: "env", Key: key, Value: jsonVal,
-		Source: zenmoney.Source, Confidence: 1.0,
-	}); err != nil {
-		return fmt.Errorf("write %s: %w", key, err)
+	if err := p.writeFact(ctx, key, zenmoney.Source, jsonVal, now); err != nil {
+		return err
 	}
 	log.Printf("mavpoll: %s read (%s)", key, zenmoney.Source)
 	return nil
@@ -506,6 +550,10 @@ func isNoFact(err error) bool {
 	return errors.Is(err, ipc.ErrNoFact)
 }

+// maxBodyBytes caps what a source can make the poller hold. Kuma's whole
+// metrics page is a few hundred kilobytes, so 4 MiB is slack, not a budget.
+const maxBodyBytes = 4 << 20
+
 func (p *poller) get(ctx context.Context, url, basicUser string) ([]byte, error) {
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
@@ -519,7 +567,7 @@ func (p *poller) get(ctx context.Context, url, basicUser string) ([]byte, error)
 		return nil, err
 	}
 	defer resp.Body.Close()
-	body, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
+	body, err := io.ReadAll(io.LimitReader(resp.Body, maxBodyBytes))
 	if err != nil {
 		return nil, err
 	}
@@ -161,6 +161,18 @@ func (s *Session) finish() error {
 	return s.spool.Sync()
 }

+// discard closes the spool and deletes it, leaving nothing behind. Used by the
+// reaper and by Abort, which throw a recording away rather than harvest it.
+func (s *Session) discard() {
+	s.mu.Lock()
+	_ = s.finish()
+	path := s.path
+	s.mu.Unlock()
+	if path != "" {
+		_ = os.Remove(path)
+	}
+}
+
 // Duration is how much audio has been collected, from the bytes rather than the
 // wall clock: a stream that dropped frames should report the audio that exists,
 // not the time that passed.
@@ -174,9 +186,16 @@ func (s *Session) duration() time.Duration {
 	return pcmDuration(s.format, s.n)
 }

+// bytesPerSample is one sample across all channels. Cutting a buffer anywhere
+// that is not a multiple of it shifts every following sample by a byte.
+func bytesPerSample(f audio.Format) int64 { return int64(f.SampleBits / 8 * f.Channels) }
+
+// bytesPerSecond is the format's byte rate, 32000 for the canonical 16 kHz mono.
+func bytesPerSecond(f audio.Format) int64 { return int64(f.SampleRate) * bytesPerSample(f) }
+
 // pcmDuration is how long n bytes of PCM lasts in the given format.
 func pcmDuration(f audio.Format, n int64) time.Duration {
-	per := int64(f.SampleRate) * int64(f.Channels) * int64(f.SampleBits) / 8
+	per := bytesPerSecond(f)
 	if per <= 0 {
 		return 0
 	}
@@ -279,19 +298,21 @@ func (r *Recorder) Start(label string) (*Session, error) {
 		f.Close()
 		return nil, err
 	}
+	// A session that never opened leaves no spool file behind.
+	abandon := func(err error) (*Session, error) {
+		f.Close()
+		_ = os.Remove(f.Name())
+		return nil, err
+	}
 	// The header is written first and rewritten at Stop with the real length,
 	// so the spool file is a playable WAV rather than headerless PCM that has
 	// to be copied to gain 44 bytes.
 	if _, err := f.Write(hdr); err != nil {
-		f.Close()
-		_ = os.Remove(f.Name())
-		return nil, fmt.Errorf("capture: spool header: %w", err)
+		return abandon(fmt.Errorf("capture: spool header: %w", err))
 	}
 	token, err := newToken()
 	if err != nil {
-		f.Close()
-		_ = os.Remove(f.Name())
-		return nil, err
+		return abandon(err)
 	}
 	s := &Session{
 		Label:   strings.TrimSpace(label),
@@ -330,15 +351,11 @@ func (r *Recorder) reapLocked() {
 	}
 	s.mu.Lock()
 	s.expired = true
-	_ = s.finish()
-	path := s.path
 	s.mu.Unlock()
-	if path != "" {
-		// The audio goes with it. A recording nobody stopped is one nobody is
-		// waiting for, and keeping it would mean storing a meeting on the
-		// strength of a dropped connection.
-		_ = os.Remove(path)
-	}
+	// The audio goes with it. A recording nobody stopped is one nobody is
+	// waiting for, and keeping it would mean storing a meeting on the strength
+	// of a dropped connection.
+	s.discard()
 	r.current = nil
 }

@@ -536,13 +553,7 @@ func (r *Recorder) Abort(token string) bool {
 		return false
 	}
 	r.current = nil
-	s.mu.Lock()
-	_ = s.finish()
-	path := s.path
-	s.mu.Unlock()
-	if path != "" {
-		_ = os.Remove(path)
-	}
+	s.discard()
 	return true
 }

@@ -576,7 +587,7 @@ func (r *Recorder) transcribeFile(ctx context.Context, path string, format audio
 		}
 		// Never cut mid-sample: a split inside an int16 shifts every following
 		// sample by a byte and turns the tail of the window into noise.
-		if bps := int64(format.SampleBits / 8 * format.Channels); bps > 0 {
+		if bps := bytesPerSample(format); bps > 0 {
 			size -= size % bps
 		}
 		if size <= 0 {
@@ -608,12 +619,13 @@ func (r *Recorder) transcribeFile(ctx context.Context, path string, format audio
 // is read by him in a note next to the words around it.
 const gapMarker = "[…не разобрала…]"

-// windowBytes is how many PCM bytes one STT window holds.
+// windowBytes is how many PCM bytes one STT window holds, rounded down to a
+// whole sample.
 func windowBytes(f audio.Format, window time.Duration) int64 {
-	bps := int64(f.SampleBits / 8 * f.Channels)
+	bps := bytesPerSample(f)
 	if bps <= 0 || f.SampleRate <= 0 || window <= 0 {
 		return 0
 	}
-	per := int64(window.Seconds()) * int64(f.SampleRate) * bps
+	per := int64(window.Seconds()) * bytesPerSecond(f)
 	return per - per%bps
 }