Recovery: reindex interrupted sessions and fix packet counts; actionable AP activation error

This commit is contained in:
Andy Dixon
2026-10-09 18:22:56 +01:00
parent 43cbcc4076
commit d8bfcee2fc
4 changed files with 72 additions and 4 deletions
+1 -1
View File
@@ -309,7 +309,7 @@ func activate(ctx context.Context, n config.Network, files map[string][]byte, st
} }
if n.AP.Enabled { if n.AP.Enabled {
if _, err := run(ctx, 60*time.Second, "nmcli", "--wait", "45", "connection", "up", netgen.ConnName); err != nil { if _, err := run(ctx, 60*time.Second, "nmcli", "--wait", "45", "connection", "up", netgen.ConnName); err != nil {
return err return fmt.Errorf("the access point did not start. Likely causes: the band/channel is not permitted for country %s or not supported by this radio, or the radio does not support %s in AP mode (%w)", n.AP.Country, strings.ToUpper(n.AP.Security), err)
} }
step("access point %q activated", n.AP.SSID) step("access point %q activated", n.AP.SSID)
if _, err := run(ctx, 30*time.Second, "systemctl", "restart", "pilens-dnsmasq.service"); err != nil { if _, err := run(ctx, 30*time.Second, "systemctl", "restart", "pilens-dnsmasq.service"); err != nil {
+14 -2
View File
@@ -130,7 +130,9 @@ func serve(args []string) error {
lim := capture.DefaultLimits() lim := capture.DefaultLimits()
db.GetJSON("capture_limits", &lim) db.GetJSON("capture_limits", &lim)
eng.SetLimits(lim) eng.SetLimits(lim)
recoverSessions(db, *caps) for _, id := range recoverSessions(db, *caps) {
go eng.Reindex(id)
}
cert, fp, err := ensureCert(*data) cert, fp, err := ensureCert(*data)
if err != nil { if err != nil {
@@ -178,7 +180,8 @@ func serve(args []string) error {
} }
// recoverSessions finalises sessions left running by a crash or power loss. // recoverSessions finalises sessions left running by a crash or power loss.
func recoverSessions(db *store.Store, dir string) { // It returns the IDs of the sessions it marked interrupted.
func recoverSessions(db *store.Store, dir string) (ids []string) {
rows, err := db.Query(`SELECT s.id, g.seq, g.file, g.format FROM segments g JOIN capture_sessions s ON s.id=g.session_id WHERE g.complete=0 AND g.deleted=0`) rows, err := db.Query(`SELECT s.id, g.seq, g.file, g.format FROM segments g JOIN capture_sessions s ON s.id=g.session_id WHERE g.complete=0 AND g.deleted=0`)
if err == nil { if err == nil {
type seg struct { type seg struct {
@@ -216,7 +219,16 @@ func recoverSessions(db *store.Store, dir string) {
slog.Warn("recovered interrupted segment", "session", s.id, "segment", s.seq, "packets", n, "truncated_tail", scanErr != nil) slog.Warn("recovered interrupted segment", "session", s.id, "segment", s.seq, "packets", n, "truncated_tail", scanErr != nil)
} }
} }
if r, err := db.Query(`SELECT id FROM capture_sessions WHERE state IN ('running','importing')`); err == nil {
for r.Next() {
var id string
r.Scan(&id)
ids = append(ids, id)
}
r.Close()
}
db.Exec(`UPDATE capture_sessions SET state='failed', error='interrupted: the service stopped unexpectedly (power loss or crash); segments were finalised on restart and summaries may be incomplete', ended_at=? WHERE state IN ('running','importing')`, time.Now().UnixMilli()) db.Exec(`UPDATE capture_sessions SET state='failed', error='interrupted: the service stopped unexpectedly (power loss or crash); segments were finalised on restart and summaries may be incomplete', ended_at=? WHERE state IN ('running','importing')`, time.Now().UnixMilli())
return ids
} }
func sdNotify(state string) { func sdNotify(state string) {
+56
View File
@@ -502,3 +502,59 @@ func (e *Engine) indexImport(id, path, format string) {
data, _ := json.Marshal(map[string]string{"session": id}) data, _ := json.Marshal(map[string]string{"session": id})
e.Hub.Publish(Event{Type: "resync", Data: data}) e.Hub.Publish(Event{Type: "resync", Data: data})
} }
// Reindex fills in summary rows missing for complete segments (after a crash or
// power loss the index can lag behind what reached the capture file).
func (e *Engine) Reindex(id string) error {
segs, err := e.Segments(id)
if err != nil {
return err
}
var total int64
for _, s := range segs {
if s.Deleted || !s.Complete {
continue
}
total += s.Count
var have int64
e.DB.QueryRow(`SELECT COUNT(*) FROM packets WHERE session_id=? AND num>=? AND num<?`, id, s.Base, s.Base+s.Count).Scan(&have)
if have >= s.Count {
continue
}
path := filepath.Join(e.sessionDir(id), s.File)
var offs []int64
pcapfile.Scan(path, s.Format, func(_ int, off int64, _ []byte) error { offs = append(offs, off); return nil }) // truncated tail tolerated
ctx, cancel := context.WithTimeout(context.Background(), time.Hour)
cmd := e.Tools.Command(ctx, 3600, e.Tools.SummaryArgs(path)...)
out, err := cmd.StdoutPipe()
if err == nil {
err = cmd.Start()
}
if err != nil {
cancel()
return err
}
var batch []Row
flush := func() {
if tx, err := e.DB.Begin(); err == nil {
insertRows(tx, id, batch)
tx.Commit()
}
batch = batch[:0]
}
ReadSummaries(out, func(r Row) {
if r.Num < 1 || r.Num > int64(len(offs)) {
return
}
r.Off, r.Seg, r.Num = offs[r.Num-1], s.Seq, s.Base+r.Num-1
if batch = append(batch, r); len(batch) >= 2000 {
flush()
}
})
flush()
cmd.Wait()
cancel()
}
e.DB.Exec(`UPDATE capture_sessions SET packets=? WHERE id=? AND packets<?`, total, id, total)
return nil
}
+1 -1
View File
@@ -91,7 +91,7 @@ func DnsmasqConf(n config.Network) string {
w := func(f string, v ...any) { fmt.Fprintf(&b, f+"\n", v...) } w := func(f string, v ...any) { fmt.Fprintf(&b, f+"\n", v...) }
w("# Managed by PiLens. Changes are overwritten; use the PiLens UI.") w("# Managed by PiLens. Changes are overwritten; use the PiLens UI.")
w("interface=%s", n.AP.Interface) w("interface=%s", n.AP.Interface)
w("bind-dynamic") // only the AP interface; survives the address appearing late w("bind-dynamic") // only the AP interface; survives the address appearing late
w("except-interface=lo") // dnsmasq otherwise adds loopback implicitly w("except-interface=lo") // dnsmasq otherwise adds loopback implicitly
w("no-hosts") w("no-hosts")
w("domain-needed") w("domain-needed")