Compare commits

..
Author SHA1 Message Date
Logan CusanoandClaude Opus 5.5 5549736706 Releases: VERSION + automatic patch tags + promote; updater installs newer versions (node-26#49)
Tag release / tag (push) Successful in 8s
CI / lint (push) Successful in 8s
CI / test-op25 (push) Successful in 16s
CI / test (push) Successful in 49s
Build edge-node / build (push) Successful in 1m2s
VERSION holds MAJOR.MINOR (1.1), edited by hand. Every push to main is
tagged vMAJOR.MINOR.PATCH by the "Tag release" workflow, with PATCH counted
as an integer by scripts/next-version.sh. "Promote release" moves the major
tag (v1) to a tested release and refuses to move it backwards.

drb-update.sh: a newer release on the node's channel is now a reason to
update by itself. REF=latest follows the highest vX.Y.Z tag (test nodes);
REF=v1 follows the promoted tag (customer nodes). Versions compare per
field as integers (1.10.0 > 1.9.3, never string or float order). It never
downgrades, and an untagged target still needs a request. After an update
it writes DRB_VERSION to .env; the edge node reports it in checkin.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-09 00:34:50 -05:00
Logan CusanoandClaude Opus 5.5 6330609a75 Node: report every radio heard on a call, not just the last (server-26#304)
CI / lint (push) Successful in 8s
CI / test-op25 (push) Successful in 16s
CI / test (push) Successful in 50s
Build edge-node / build (push) Successful in 9m46s
P1 of the radio-ID plan. metadata_watcher records each transmission as
[radio, first seen, last seen] from channel rows and call_log grants in both
audio and console modes. Sightings of the same radio under 2 s apart are
merged, because the P0 probe found the ID in only 44% of polls during a
burst. call_end gains `radios: [{radio, start, end}]`.

main.py rebases the offsets into the trimmed audio file's frame, so
wall clock = audio_start_epoch + offset, the mapping C2 already uses. It
also sends the list on /upload (UploadMeta.radios), because C2 starts the
pipeline from /upload before call_end arrives.

srcaddr keeps its meaning (latest talker). Metadata only: no boundary,
timestamp or attribution change (drb-audio-review).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-08 23:03:18 -05:00
logan c57a5b9068 Merge pull request 'Upload spool + retry, analog noise squelch, MQTT error, Discord pause (node-26#44 #46 #21 #34)' (#47) from fix/44-upload-spool into main
CI / test-op25 (push) Successful in 19s
CI / lint (push) Successful in 7s
CI / test (push) Successful in 55s
Build edge-node / build (push) Successful in 11m4s
Build op25 / build (push) Successful in 1h41m54s
Merge pull request 'Upload spool + retry, analog noise squelch (node-26#44 #46 #21 #34)' (#47) from fix/44-upload-spool into main
2026-11-01 06:39:39 -05:00
14 changed files with 406 additions and 10 deletions
+39
View File
@@ -0,0 +1,39 @@
name: Promote release
# node-26#49: move the major tag (v1) to a tested release. Customer nodes
# (updater REF=v1) install it on their next timer run, because its version is
# numerically higher than what they run. Never moves it backwards.
on:
workflow_dispatch:
inputs:
version:
description: "Release to promote, e.g. 1.1.3 (must already be tagged v1.1.3)"
required: true
jobs:
promote:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Move the major tag
env:
VERSION: ${{ inputs.version }}
run: |
[[ "$VERSION" =~ ^([0-9]+)\.[0-9]+\.[0-9]+$ ]] || { echo "version must be MAJOR.MINOR.PATCH, got '$VERSION'"; exit 1; }
MAJOR="${BASH_REMATCH[1]}"
git fetch --force --tags origin
git rev-parse -q --verify "refs/tags/v$VERSION" >/dev/null || { echo "tag v$VERSION does not exist"; exit 1; }
# Refuse to go backwards: compare with the release v$MAJOR points at now.
CUR="$(git tag --points-at "v$MAJOR" 2>/dev/null | grep -E '^v[0-9]+\.[0-9]+\.[0-9]+$' \
| sed 's/^v//' | sort -t. -k1,1n -k2,2n -k3,3n | tail -1 || true)"
if [ -n "$CUR" ]; then
HIGHEST="$(printf '%s\n%s\n' "$CUR" "$VERSION" | sort -t. -k1,1n -k2,2n -k3,3n | tail -1)"
[ "$HIGHEST" = "$VERSION" ] || { echo "v$MAJOR is at $CUR, newer than $VERSION — refusing to roll back here"; exit 1; }
fi
git tag -f "v$MAJOR" "v$VERSION^{commit}"
git push --force origin "refs/tags/v$MAJOR"
echo "v$MAJOR -> $VERSION (was ${CUR:-untagged})"
+31
View File
@@ -0,0 +1,31 @@
name: Tag release
# node-26#49: every push to main is a release, vMAJOR.MINOR.PATCH (see
# scripts/next-version.sh). Test nodes (updater REF=latest) install the
# highest tag on their next timer run; customer nodes (REF=v1) wait for the
# Promote workflow.
on:
push:
branches: [main]
concurrency:
group: version-tag
cancel-in-progress: false
jobs:
tag:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Tag this commit
run: |
git fetch --force --tags origin
TAG="$(bash scripts/next-version.sh)"
if [ -z "$TAG" ]; then echo "already tagged"; exit 0; fi
git tag "$TAG"
git push origin "refs/tags/$TAG"
echo "released $TAG"
+1
View File
@@ -0,0 +1 @@
1.1
+3
View File
@@ -6,6 +6,9 @@ class Settings(BaseSettings):
# Node identity
node_id: str
node_name: str = "Unnamed Node"
# The release installed on this node (node-26#49). The host updater writes
# DRB_VERSION to .env after each update; "untagged" before the first one.
drb_version: str = "untagged"
node_lat: float = 0.0
node_lon: float = 0.0
@@ -785,6 +785,7 @@ class CallRecorder:
system_id: Optional[str] = None,
audio_start_epoch: Optional[float] = None,
audio_end_epoch: Optional[float] = None,
radios: Optional[str] = None,
) -> Optional[str]:
"""
Upload once, inline. Returns the storage URL on success.
@@ -802,6 +803,7 @@ class CallRecorder:
meta = UploadMeta(
call_id=call_id, talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name,
system_id=system_id, audio_start_epoch=audio_start_epoch, audio_end_epoch=audio_end_epoch,
radios=radios,
)
result = await self._post_inline(file_path, meta)
self.spool.last_result = {"call_id": call_id, "kind": result.kind, "status": result.status,
@@ -132,6 +132,11 @@ ATTRIBUTION_LOOKAHEAD_SECONDS = 2.0
CONSOLE_HISTORY_SECONDS = 180.0
CONSOLE_HISTORY_MAX = 1200
# Sightings of the same radio closer together than this are one transmission.
# The radio ID shows up in only ~44% of 0.5 s polls during a burst (server-26#304
# P0 probe, 2026-11-08), so gaps of a few polls inside one key-up are normal.
TX_MERGE_GAP_SECONDS = 2.0
# How far back to look for a duplicate before appending a grant. OP25's call_log
# deque drains on read so repeats should not happen, but a re-delivered entry
# would otherwise inflate the transmission count and the attribution score.
@@ -207,6 +212,21 @@ def _talker(channel: Dict[str, Any]) -> Optional[tuple]:
return tgid, rid
def rebase_radios(radios: List[Dict[str, Any]], shift: float, span: float) -> List[Dict[str, Any]]:
"""Move call-relative radio offsets into another frame (the trimmed audio file).
`shift` is call start minus the new origin; offsets land clamped to [0, span]
so a talker before the file's first sample sits at 0, not before it.
"""
span = max(0.0, span)
return [
{**r,
"start": round(min(max(0.0, r["start"] + shift), span), 2),
"end": round(min(max(0.0, r["end"] + shift), span), 2)}
for r in radios
]
def _iso(epoch: Optional[float]) -> Optional[str]:
if epoch is None:
return None
@@ -254,6 +274,10 @@ class MetadataWatcher:
self._current_srcaddr: Optional[int] = None
self._started_at: Optional[float] = None # audio onset, or grant epoch in fallback mode
self._transmissions: int = 0
# Every radio heard on the open segment, in order: [rid, first_epoch,
# last_epoch]. srcaddr above keeps only the latest one; one recording
# often holds a dispatcher and a unit (server-26#304).
self._tx: List[List[Any]] = []
# True when the open segment is governed by audio, False for the
# console fallback. Fixed at open so capture flapping cannot switch the
# rules underneath a live segment.
@@ -559,6 +583,7 @@ class MetadataWatcher:
self._current_tgid_name = entry.get("tgtag") or ""
self._current_freq = entry.get("freq")
self._current_srcaddr = _as_int(entry.get("rid"))
self._note_tx(self._current_srcaddr, started_at)
self._transmissions += 1
logger.info(
f"Late attribution: segment {self._active_call_id} adopted tgid {tgid} from a grant "
@@ -571,11 +596,13 @@ class MetadataWatcher:
# of a single conversation lands in one file.
self._transmissions += 1
self._refresh_meta_from_log(entry)
self._note_tx(_as_int(entry.get("rid")), started_at)
return
if self._same_channel(tgid, entry.get("freq"), entry.get("tgtag") or ""):
self._transmissions += 1
self._refresh_meta_from_log(entry)
self._note_tx(_as_int(entry.get("rid")), started_at)
return
# FORCED SPLIT. Two talkgroups can be back to back with no silence
@@ -626,6 +653,7 @@ class MetadataWatcher:
ours = True
if srcaddr is not None:
self._current_srcaddr = srcaddr
self._note_tx(srcaddr, now)
self._last_activity = now
self._refresh_meta_from_channel(channel)
@@ -638,6 +666,7 @@ class MetadataWatcher:
self._current_tgid_name = active[0].get("tag") or ""
self._current_freq = active[0].get("freq")
self._current_srcaddr = _as_int(active[0].get("srcaddr"))
self._note_tx(self._current_srcaddr, now)
logger.info(f"Late attribution: segment {self._active_call_id} adopted tgid {tgid} from channel state.")
return
@@ -709,6 +738,7 @@ class MetadataWatcher:
self._last_tx_end = None
self._last_activity = now
self._refresh_meta_from_log(entry)
self._note_tx(_as_int(entry.get("rid")), started_at)
return
await self._close_segment(started_at + _tail_pad(), reason="tgid_change")
@@ -730,6 +760,7 @@ class MetadataWatcher:
tx_active = True
if srcaddr is not None:
self._current_srcaddr = srcaddr
self._note_tx(srcaddr, now)
self._refresh_meta_from_channel(channel)
elif chan_tgid is not None:
foreign_active_tgid = chan_tgid
@@ -792,6 +823,28 @@ class MetadataWatcher:
# Segment open / close
# ------------------------------------------------------------------
def _note_tx(self, rid: Optional[int], epoch: float) -> None:
"""Record a sighting of `rid` talking on the open segment."""
if rid is None or self._active_call_id is None:
return
last = self._tx[-1] if self._tx else None
if last is not None and last[0] == rid and epoch - last[2] <= TX_MERGE_GAP_SECONDS:
last[2] = max(last[2], epoch)
return
self._tx.append([rid, epoch, epoch])
def _radios(self, start: float, end: float) -> List[Dict[str, Any]]:
"""The segment's transmissions as offsets (seconds) from its start, clamped to it."""
span = max(0.0, end - start)
return [
{
"radio": rid,
"start": round(min(max(0.0, first - start), span), 2),
"end": round(min(max(0.0, last - start), span), 2),
}
for rid, first, last in self._tx
]
def _refresh_meta_from_log(self, entry: Dict[str, Any]) -> None:
if not self._current_tgid_name:
self._current_tgid_name = entry.get("tgtag") or ""
@@ -826,6 +879,8 @@ class MetadataWatcher:
self._started_at = started_at
self._transmissions = max(1, transmissions)
self._audio_driven = audio_driven
self._tx = []
self._note_tx(srcaddr, started_at)
# Console fallback assumes the transmission is still up; it learns
# otherwise from the next channel scan.
@@ -876,6 +931,7 @@ class MetadataWatcher:
"ended_at": _iso(end_epoch),
"ended_at_epoch": end_epoch,
"transmissions": self._transmissions,
"radios": self._radios(started_at if started_at is not None else end_epoch, end_epoch),
"end_reason": reason,
"attributed": attributed,
"driver": "audio" if self._audio_driven else "console",
@@ -910,6 +966,7 @@ class MetadataWatcher:
self._current_srcaddr = None
self._started_at = None
self._transmissions = 0
self._tx = []
self._tx_active = False
self._last_tx_end = None
self._audio_driven = False
@@ -949,6 +1006,10 @@ class MetadataWatcher:
self._current_freq = found.freq
if found.rid is not None and self._current_srcaddr is None:
self._current_srcaddr = found.rid
if not self._tx:
# Only as the sole entry: the window's latest radio has no
# timing of its own, so it must not reorder real sightings.
self._note_tx(found.rid, start)
self._transmissions = max(self._transmissions, found.grants)
return
@@ -378,6 +378,7 @@ class MQTTManager:
"enforce_override_timeout": config.enforce_override_timeout,
"secondary_sdr_mode": config.secondary_sdr_mode,
"secondary_sdr_priority": config.secondary_sdr_priority,
"version": settings.drb_version,
}
payload["sdr_pins"] = config.sdr_pins
# Conventional channels (node-26#24): name/freq/NAC and whether the
@@ -71,6 +71,10 @@ class UploadMeta:
system_id: Optional[str] = None
audio_start_epoch: Optional[float] = None
audio_end_epoch: Optional[float] = None
# JSON list of {radio, start, end} — every radio heard on the call
# (server-26#304). Sent here as well as on call_end because C2 starts the
# pipeline from /upload, before call_end arrives.
radios: Optional[str] = None
@dataclass
@@ -108,6 +112,8 @@ def _form(meta: UploadMeta) -> Dict[str, str]:
form["audio_start_epoch"] = f"{meta.audio_start_epoch:.3f}"
if meta.audio_end_epoch is not None:
form["audio_end_epoch"] = f"{meta.audio_end_epoch:.3f}"
if meta.radios:
form["radios"] = meta.radios
return form
+12 -1
View File
@@ -1,4 +1,5 @@
import asyncio
import json
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Optional
@@ -9,7 +10,7 @@ from app.models import SystemConfig, normalize_sdr_pins, normalize_secondary_pri
from app.internal.logger import logger
from app.internal.mqtt_manager import mqtt_manager
from app.internal import credentials
from app.internal.metadata_watcher import metadata_watcher
from app.internal.metadata_watcher import metadata_watcher, rebase_radios
from app.internal.call_recorder import call_recorder
from app.internal.discord_radio import radio_bot
from app.internal.config_manager import (
@@ -93,6 +94,15 @@ async def on_call_end(data: dict):
data["audio_end_at"] = _iso(recording.audio_end_epoch)
data["audio_start_epoch"] = recording.audio_start_epoch
data["audio_end_epoch"] = recording.audio_end_epoch
# Radio offsets were measured from the call start; C2 maps an offset
# to wall clock as audio_start_epoch + t (above), so put them in the
# audio file's frame (server-26#304).
if data.get("radios") and data.get("started_at_epoch") is not None:
data["radios"] = rebase_radios(
data["radios"],
shift=data["started_at_epoch"] - recording.audio_start_epoch,
span=recording.audio_end_epoch - recording.audio_start_epoch,
)
data["audio_lead_trimmed"] = round(recording.lead_trimmed, 3)
data["audio_tail_trimmed"] = round(recording.tail_trimmed, 3)
if recording.clamped_seconds:
@@ -108,6 +118,7 @@ async def on_call_end(data: dict):
system_id=node_cfg.assigned_system_id,
audio_start_epoch=recording.audio_start_epoch,
audio_end_epoch=recording.audio_end_epoch,
radios=json.dumps(data["radios"]) if data.get("radios") else None,
)
if audio_url:
data["audio_url"] = audio_url
+77 -1
View File
@@ -44,6 +44,11 @@ case "$1" in
*:docker-compose.yml) cat "$STUB_REPO/docker-compose.yml" ;;
esac ;;
checkout) echo "${@: -1}" > "$STUB/head" ;;
tag)
if [ "$2" = --points-at ]; then
if [ "$3" = HEAD ]; then cat "$STUB/tags_head" 2>/dev/null; else cat "$STUB/tags_new" 2>/dev/null; fi
else cat "$STUB/tags_all" 2>/dev/null; fi
exit 0 ;;
*) exit 0 ;;
esac
'''
@@ -121,7 +126,7 @@ def test_no_request_means_checkin_only(env):
calls = _calls(env)
assert "update-check" in calls
assert "docker" not in calls
assert "fetch" not in calls
assert "checkout" not in calls
def test_c2_request_updates_once_and_reports(env):
@@ -248,3 +253,74 @@ def test_target_without_pinning_still_updates(env):
assert (env["stub"] / "head").read_text().strip() == NEW
assert _pulled(_calls(env))
assert "DRB_TAG_" not in (env["inst"] / ".env").read_text()
# ---------------------------------------------------------------------------
# Releases (node-26#49): a newer version on the node's channel is a reason to
# update by itself, and versions compare per field as integers.
# ---------------------------------------------------------------------------
def _channel(env, ref, all_tags, target_tags, head_tags):
env["conf"].write_text(env["conf"].read_text().replace("REF=v1", f"REF={ref}"))
(env["stub"] / "tags_all").write_text("\n".join(all_tags) + "\n")
(env["stub"] / "tags_new").write_text("\n".join(target_tags) + "\n")
(env["stub"] / "tags_head").write_text("\n".join(head_tags) + "\n")
def test_latest_channel_installs_a_newer_release_unasked(env):
# 1.1.10 is newer than 1.1.9: compared as integers, not strings or floats.
_channel(env, "latest", ["v1", "v1.1.9", "v1.1.10", "v1.1.2"], ["v1.1.10"], ["v1.1.9"])
r = _run(env)
assert r.returncode == 0, r.stderr + r.stdout
assert "release 1.1.10 on latest (installed 1.1.9)" in r.stdout
assert (env["stub"] / "head").read_text().strip() == NEW
assert "DRB_VERSION=1.1.10" in (env["inst"] / ".env").read_text()
def test_never_downgrades(env):
_channel(env, "latest", ["v1.1.9", "v1.2.0"], ["v1.1.9"], ["v1.2.0"])
r = _run(env)
assert r.returncode == 0, r.stderr
assert "docker" not in _calls(env)
def test_same_release_is_not_reinstalled(env):
_channel(env, "v1", ["v1", "v1.1.3"], ["v1", "v1.1.3"], ["v1", "v1.1.3"])
assert _run(env).returncode == 0
assert "docker" not in _calls(env)
def test_promoted_major_tag_installs_on_stable_nodes(env):
_channel(env, "v1", ["v1", "v1.1.0", "v1.1.4"], ["v1", "v1.1.4"], ["v1.1.0"])
r = _run(env)
assert r.returncode == 0, r.stderr + r.stdout
assert "release 1.1.4 on v1" in r.stdout
assert (env["stub"] / "head").read_text().strip() == NEW
def test_untagged_target_still_needs_a_request(env):
"""v1 before its first promotion points at a commit with no release tag."""
_channel(env, "v1", ["v1"], ["v1"], [])
assert _run(env).returncode == 0
assert "docker" not in _calls(env)
def test_next_version_counts_patches_as_integers(tmp_path):
script = REPO_ROOT / "scripts" / "next-version.sh"
g = ["git", "-c", "user.email=t@t", "-c", "user.name=t", "-C", str(tmp_path)]
subprocess.run(["git", "init", "-q", str(tmp_path)], check=True)
(tmp_path / "VERSION").write_text("1.1\n")
subprocess.run(g + ["commit", "-q", "--allow-empty", "-m", "a"], check=True)
def nxt():
return subprocess.run(["bash", str(script)], cwd=tmp_path, capture_output=True, text=True,
check=True).stdout.strip()
assert nxt() == "v1.1.0"
for t in ("v1.1.0", "v1.1.9", "v1.0.30"):
subprocess.run(g + ["tag", t], check=True)
assert nxt() == "", "a commit that already has a release gets no second one"
subprocess.run(g + ["commit", "-q", "--allow-empty", "-m", "b"], check=True)
assert nxt() == "v1.1.10"
(tmp_path / "VERSION").write_text("1.2\n")
assert nxt() == "v1.2.0"
@@ -27,6 +27,7 @@ from app.internal.metadata_watcher import (
MAX_SEGMENT_SECONDS,
MetadataWatcher,
OP25_OFFLINE_GRACE,
rebase_radios,
)
from app.internal.op25_client import TerminalUpdate, parse_terminal_messages
@@ -1147,3 +1148,86 @@ async def test_console_history_is_bounded(hearing, clock, audio):
clock.advance(CONSOLE_HISTORY_SECONDS + 1)
await tick(hearing, update())
assert all(e.epoch >= clock.now - CONSOLE_HISTORY_SECONDS for e in hearing._console)
# ---------------------------------------------------------------------------
# Every radio on the call (server-26#304) — not just the last srcaddr
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_call_end_lists_every_radio_in_order(watcher, clock):
"""Dispatcher, unit, dispatcher on one talkgroup: three transmissions."""
start = clock.now
await tick(watcher, update(call_log=[grant(1234, start, rid=555)],
channels=[channel(tgid=1234, srcaddr=555)]))
clock.advance(1.0)
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)]))
clock.advance(1.0)
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=777)]))
clock.advance(0.5)
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=777)]))
clock.advance(1.0)
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)]))
clock.advance(0.5)
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
clock.advance(settings.call_idle_timeout + 0.5)
await tick(watcher, update(channels=[channel()]))
payload = watcher.on_call_end.call_args[0][0]
assert [r["radio"] for r in payload["radios"]] == [555, 777, 555]
assert payload["radios"][0] == {"radio": 555, "start": 0.0, "end": 1.0}
assert payload["radios"][1] == {"radio": 777, "start": 2.0, "end": 2.5}
# srcaddr keeps its old meaning (the latest talker) for existing readers.
assert payload["srcaddr"] == 555
@pytest.mark.asyncio
async def test_missed_polls_inside_one_keyup_stay_one_transmission(watcher, clock):
"""The radio ID drops out of some polls mid-transmission; that is not a new key-up."""
await tick(watcher, update(call_log=[grant(1234, clock.now, rid=555)],
channels=[channel(tgid=1234, srcaddr=555)]))
clock.advance(0.5)
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
clock.advance(1.0)
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)]))
clock.advance(0.5)
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
clock.advance(settings.call_idle_timeout + 0.5)
await tick(watcher, update(channels=[channel()]))
radios = watcher.on_call_end.call_args[0][0]["radios"]
assert radios == [{"radio": 555, "start": 0.0, "end": 1.5}]
@pytest.mark.asyncio
async def test_radio_list_resets_between_calls(watcher, clock):
await tick(watcher, update(call_log=[grant(1234, clock.now, rid=555)]))
clock.advance(0.2)
await tick(watcher, update(call_log=[grant(4321, clock.now, rid=888)]))
first = watcher.on_call_end.call_args[0][0]
assert [r["radio"] for r in first["radios"]] == [555]
assert watcher._tx == [[888, clock.now, clock.now]]
@pytest.mark.asyncio
async def test_audio_mode_records_a_radio_seen_only_in_a_grant(hearing, clock, audio):
"""A short key-up between channel polls is only visible as a call_log grant."""
audio.speak()
await tick(hearing, update(call_log=[grant(1234, clock.now, rid=555)],
channels=[channel(tgid=1234, srcaddr=555)]))
clock.advance(1.0)
audio.speak()
await tick(hearing, update(call_log=[grant(1234, clock.now, rid=777)]))
clock.advance(settings.call_silence_timeout + 0.5)
await tick(hearing, update(channels=[channel()]))
radios = hearing.on_call_end.call_args[0][0]["radios"]
assert [r["radio"] for r in radios] == [555, 777]
def test_rebase_radios_moves_offsets_into_the_audio_frame():
"""The file starts 0.25 s of pre-roll before the call, then 1.0 s of lead was trimmed."""
radios = [{"radio": 555, "start": 0.0, "end": 2.0}, {"radio": 777, "start": 3.0, "end": 9.0}]
out = rebase_radios(radios, shift=-0.75, span=6.0)
assert out == [{"radio": 555, "start": 0.0, "end": 1.25}, {"radio": 777, "start": 2.25, "end": 6.0}]
+7
View File
@@ -228,3 +228,10 @@ async def test_wake_makes_everything_due_now(tmp_path):
sp.add(_recording(tmp_path), UploadMeta(call_id="c1"), UploadResult(RETRY, status=401))
sp.wake()
assert await sp.drain_once() == 1
def test_form_carries_the_radio_list_and_omits_it_when_empty():
"""C2 starts correlation from /upload, so the radio list rides on it (server-26#304)."""
radios = json.dumps([{"radio": 555, "start": 0.0, "end": 1.0}])
assert upload_spool._form(UploadMeta(call_id="c1", radios=radios))["radios"] == radios
assert "radios" not in upload_spool._form(UploadMeta(call_id="c1"))
+62 -8
View File
@@ -1,7 +1,10 @@
#!/usr/bin/env bash
#
# DRB node host updater (node-26#15). Runs as root from systemd:
# drb-update.timer -> every ~15 min: HTTPS check-in with C2; update if asked
# drb-update.timer -> every ~15 min: HTTPS check-in with C2; update if asked,
# or if the configured channel has a newer release
# (node-26#49: REF=latest follows the highest vX.Y.Z
# tag, REF=v1 follows the promoted major tag)
# drb-update.path -> fires when the edge-node container drops
# <install>/configs/update-requested (MQTT node_update)
# Both start drb-update.service, which runs this script. install-updater.sh
@@ -26,6 +29,42 @@
# very file mid-run cannot change what is executing.
set -euo pipefail
# ── Release versions (node-26#49) ─────────────────────────────────────────
# Versions are MAJOR.MINOR.PATCH and every comparison is per field, as
# integers: 1.10.0 > 1.9.3. Never string or float order ("1.10" < "1.9" as a
# float, "1.10.0" < "1.9.3" as a string).
semver_sort() { sort -t. -k1,1n -k2,2n -k3,3n; }
semver_gt() { # semver_gt A B -> true when A is newer than B
local a b i
IFS=. read -r -a a <<< "$1"; IFS=. read -r -a b <<< "$2"
for i in 0 1 2; do
(( 10#${a[i]:-0} > 10#${b[i]:-0} )) && return 0
(( 10#${a[i]:-0} < 10#${b[i]:-0} )) && return 1
done
return 1
}
version_at() { # highest X.Y.Z release tagged on a commit, or nothing
git tag --points-at "$1" 2>/dev/null | grep -E '^v[0-9]+\.[0-9]+\.[0-9]+$' \
| sed 's/^v//' | semver_sort | tail -1 || true
}
resolve_target() { # REF -> a ref name git can resolve, or fail
local ref="$1" top
if [ "$ref" = latest ]; then
top="$(git tag -l 'v*' | grep -E '^v[0-9]+\.[0-9]+\.[0-9]+$' | sed 's/^v//' | semver_sort | tail -1 || true)"
[ -n "$top" ] || return 1
echo "refs/tags/v$top"
elif git show-ref --verify -q "refs/remotes/origin/$ref"; then
echo "refs/remotes/origin/$ref"
elif git show-ref --verify -q "refs/tags/$ref"; then
echo "refs/tags/$ref"
else
return 1
fi
}
main() {
local CONF="${DRB_UPDATER_CONF:-/etc/drb/updater.conf}"
local STATE_DIR="${DRB_UPDATER_STATE:-/var/lib/drb-update}"
@@ -118,6 +157,22 @@ main() {
new_request="$requested"
reason="${reason:-C2 request $requested}"
fi
# A newer release on this node's channel is a reason on its own (node-26#49).
# Only ever forward, and only between tagged releases: an untagged target
# (a branch REF, or v1 before its first promotion) still needs a request.
if [ -z "$reason" ]; then
local origin_url target tv iv
origin_url="$(git remote get-url origin 2>/dev/null || true)"
if [ "$origin_url" = "$REPO_URL" ] \
&& git -c protocol.allow=never -c protocol.https.allow=always fetch -q --force --tags --prune origin \
&& target="$(resolve_target "$REF")"; then
tv="$(version_at "$(git rev-parse "$target^{commit}")")"
iv="$(version_at HEAD)"
if [ -n "$tv" ] && { [ -z "$iv" ] || semver_gt "$tv" "$iv"; }; then
reason="release $tv on $REF (installed ${iv:-untagged})"
fi
fi
fi
[ -n "$reason" ] || { log "no update requested"; return 0; }
# ── Update ─────────────────────────────────────────────────────────────
@@ -145,13 +200,7 @@ do_update() {
git -c protocol.allow=never -c protocol.https.allow=always fetch --force --tags --prune origin || return 1
local target
if git show-ref --verify -q "refs/remotes/origin/$REF"; then
target="refs/remotes/origin/$REF"
elif git show-ref --verify -q "refs/tags/$REF"; then
target="refs/tags/$REF"
else
err "ref '$REF' not found on origin"; return 1
fi
target="$(resolve_target "$REF")" || { err "ref '$REF' not found on origin"; return 1; }
local old new
old="$(git rev-parse HEAD)"; new="$(git rev-parse "$target^{commit}")"
@@ -206,6 +255,11 @@ do_update() {
drb_pin_image_tags "$INSTALL_DIR/.env" "$INSTALL_DIR" "$new" || { err "could not write image pins to .env"; return 1; }
fi
# The release this node now runs, for its checkin (node-26#49).
local ver; ver="$(version_at "$new")"
sed -i '/^DRB_VERSION=/d' "$INSTALL_DIR/.env"
printf 'DRB_VERSION=%s\n' "${ver:-untagged}" >> "$INSTALL_DIR/.env"
if [ "$BUILD" = 1 ]; then
docker compose build || return 1 # tags the local builds with the same pins
fi
+20
View File
@@ -0,0 +1,20 @@
#!/usr/bin/env bash
# node-26#49 — the release tag this commit should get, or nothing if it has one.
#
# VERSION holds MAJOR.MINOR and is edited by hand. PATCH is automatic: one
# more than the highest existing vMAJOR.MINOR.PATCH tag, compared as an
# integer (so .10 follows .9), starting at 0 for a new MAJOR.MINOR.
#
# Usage: scripts/next-version.sh (from the repo root, tags fetched)
set -euo pipefail
base="$(tr -d ' \r\n' < VERSION)"
[[ "$base" =~ ^[0-9]+\.[0-9]+$ ]] || { echo "VERSION must be MAJOR.MINOR, got '$base'" >&2; exit 1; }
pattern="^v${base//./\\.}\\.[0-9]+$"
if git tag --points-at HEAD | grep -qE "$pattern"; then
exit 0 # this commit is already released
fi
last="$(git tag -l "v${base}.*" | grep -E "$pattern" | sed "s/^v${base}\.//" | sort -n | tail -1 || true)"
echo "v${base}.$(( ${last:--1} + 1 ))"