Skip to content

Commit 959507b

Browse files
adubovikovadubovikov
authored
Merge pull request #911 from sipcapture/fix/909-rtpengine-date-partitions-11.0.306
fix(decoder): sanitize HEP timestamps for RTPengine (#909)
2 parents 8b9e275 + 21f9749 commit 959507b

4 files changed

Lines changed: 207 additions & 10 deletions

File tree

‎src/decoder/decoder.go‎

Lines changed: 71 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -222,11 +222,7 @@ func (h *HEP) parse(packet []byte) error {
222222
hepV3Enable, hepV2Enable, protobufEnable)
223223
}
224224

225-
h.Timestamp = time.Unix(int64(h.Tsec), int64(h.Tmsec*1000))
226-
if h.Tsec == 0 && h.Tmsec == 0 {
227-
logger.Debug("got null timestamp", "nodeID", h.NodeID)
228-
h.Timestamp = time.Now()
229-
}
225+
h.sanitizeHEPTimestamp(time.Now())
230226

231227
h.normPayload()
232228
h.HEPParseNs = time.Now().UnixNano() - hepStart
@@ -278,6 +274,76 @@ func (h *HEP) parse(packet []byte) error {
278274
return nil
279275
}
280276

277+
// ntpUnixOffset is seconds between NTP epoch (1900-01-01) and Unix epoch (1970-01-01).
278+
const ntpUnixOffset uint32 = 2208988800
279+
280+
// hepTimestampPastWindow / hepTimestampFutureWindow bound acceptable HEP wall times
281+
// relative to receive time. Far-future Tsec (NTP mis-encoded as Unix, clock skew)
282+
// otherwise creates DuckLake date= partitions decades ahead and can OOM the catalog.
283+
const (
284+
hepTimestampPastWindow = 10 * 365 * 24 * time.Hour
285+
hepTimestampFutureWindow = 24 * time.Hour
286+
)
287+
288+
// sanitizeHEPTimestamp normalizes HEP chunk 0x0009/0x000a into a sane wall clock,
289+
// writing Timestamp, Tsec, and Tmsec in sync so DuckLake date partitions stay correct.
290+
// See https://github.com/sipcapture/homer/issues/909
291+
func (h *HEP) sanitizeHEPTimestamp(now time.Time) {
292+
now = now.UTC()
293+
294+
// HEP usec must be < 1e6; larger values are typically NTP fractional seconds
295+
// wrongly placed in chunk 0x000a (not wall-clock microseconds).
296+
if h.Tmsec >= 1_000_000 {
297+
logger.Debug("invalid HEP usec, clamping to 0", "nodeID", h.NodeID, "tmsec", h.Tmsec)
298+
h.Tmsec = 0
299+
}
300+
301+
if h.Tsec == 0 {
302+
logger.Debug("got null timestamp", "nodeID", h.NodeID)
303+
h.applyTimestamp(now)
304+
return
305+
}
306+
307+
// Cast Tmsec to int64 before *1000 to avoid uint32 overflow.
308+
ts := time.Unix(int64(h.Tsec), int64(h.Tmsec)*1000).UTC()
309+
if hepTimestampInWindow(ts, now) {
310+
h.applyTimestamp(ts)
311+
return
312+
}
313+
314+
// NTP seconds misread as Unix: subtract NTP→Unix offset and re-check.
315+
if h.Tsec > ntpUnixOffset {
316+
converted := time.Unix(int64(h.Tsec-ntpUnixOffset), int64(h.Tmsec)*1000).UTC()
317+
if hepTimestampInWindow(converted, now) {
318+
logger.Debug("HEP timestamp looked like NTP epoch, converted to Unix",
319+
"nodeID", h.NodeID, "rawTsec", h.Tsec, "converted", converted)
320+
h.applyTimestamp(converted)
321+
return
322+
}
323+
}
324+
325+
logger.Debug("HEP timestamp out of range, using receive time",
326+
"nodeID", h.NodeID, "rawTsec", h.Tsec, "decoded", ts)
327+
h.applyTimestamp(now)
328+
}
329+
330+
func hepTimestampInWindow(ts, now time.Time) bool {
331+
if ts.After(now.Add(hepTimestampFutureWindow)) {
332+
return false
333+
}
334+
if ts.Before(now.Add(-hepTimestampPastWindow)) {
335+
return false
336+
}
337+
return true
338+
}
339+
340+
func (h *HEP) applyTimestamp(ts time.Time) {
341+
ts = ts.UTC()
342+
h.Timestamp = ts
343+
h.Tsec = uint32(ts.Unix())
344+
h.Tmsec = uint32(ts.Nanosecond() / 1000)
345+
}
346+
281347
func (h *HEP) normPayload() {
282348
// Check deduplicate setting from decoder config or legacy homerconfig
283349
deduplicate := false

‎src/decoder/timestamp_test.go‎

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
// Copyright (C) 2026 Homer Server Contributors
2+
//
3+
// SPDX-License-Identifier: AGPL-3.0-or-later
4+
5+
package decoder
6+
7+
import (
8+
"testing"
9+
"time"
10+
)
11+
12+
func TestSanitizeHEPTimestamp(t *testing.T) {
13+
// Fixed "receive" time matching the issue #909 sample window (2026-08-04).
14+
now := time.Date(2026, 8, 4, 12, 0, 0, 0, time.UTC)
15+
validUnix := uint32(time.Date(2026, 8, 3, 11, 33, 46, 0, time.UTC).Unix()) // 1785756826
16+
ntpAsUnix := uint32(3994745626) // NTP MSW misread as Unix → 2096
17+
brokenHeader := uint32(2608623722) // packet 490 tv_sec → 2052
18+
brokenUsec := uint32(1059783936) // NTP frac in usec field
19+
20+
tests := []struct {
21+
name string
22+
tsec uint32
23+
tmsec uint32
24+
wantTsec uint32
25+
wantTmsec uint32
26+
wantNow bool // expect receive time
27+
checkExact bool
28+
}{
29+
{
30+
name: "normal unix preserved",
31+
tsec: validUnix,
32+
tmsec: 123456,
33+
wantTsec: validUnix,
34+
wantTmsec: 123456,
35+
checkExact: true,
36+
},
37+
{
38+
name: "ntp-as-unix converted",
39+
tsec: ntpAsUnix,
40+
tmsec: 0,
41+
wantTsec: validUnix,
42+
wantTmsec: 0,
43+
checkExact: true,
44+
},
45+
{
46+
name: "broken header falls back to receive time",
47+
tsec: brokenHeader,
48+
tmsec: brokenUsec,
49+
wantNow: true,
50+
},
51+
{
52+
name: "zero timestamp uses receive time and syncs Tsec",
53+
tsec: 0,
54+
tmsec: 0,
55+
wantNow: true,
56+
},
57+
{
58+
name: "valid usec under 1e6 preserved",
59+
tsec: validUnix,
60+
tmsec: 999999,
61+
wantTsec: validUnix,
62+
wantTmsec: 999999,
63+
checkExact: true,
64+
},
65+
{
66+
name: "invalid usec clamped then out-of-range falls back",
67+
tsec: brokenHeader,
68+
tmsec: brokenUsec,
69+
wantNow: true,
70+
},
71+
}
72+
73+
for _, tt := range tests {
74+
t.Run(tt.name, func(t *testing.T) {
75+
h := &HEP{Tsec: tt.tsec, Tmsec: tt.tmsec, NodeID: 2001}
76+
h.sanitizeHEPTimestamp(now)
77+
78+
if tt.wantNow {
79+
if h.Timestamp.UTC() != now {
80+
t.Fatalf("Timestamp = %v, want receive time %v", h.Timestamp, now)
81+
}
82+
if h.Tsec != uint32(now.Unix()) {
83+
t.Fatalf("Tsec = %d, want %d (synced receive time)", h.Tsec, now.Unix())
84+
}
85+
if h.Tmsec != uint32(now.Nanosecond()/1000) {
86+
t.Fatalf("Tmsec = %d, want %d", h.Tmsec, now.Nanosecond()/1000)
87+
}
88+
return
89+
}
90+
91+
if !tt.checkExact {
92+
return
93+
}
94+
if h.Tsec != tt.wantTsec {
95+
t.Fatalf("Tsec = %d, want %d", h.Tsec, tt.wantTsec)
96+
}
97+
if h.Tmsec != tt.wantTmsec {
98+
t.Fatalf("Tmsec = %d, want %d", h.Tmsec, tt.wantTmsec)
99+
}
100+
if h.Timestamp.Unix() != int64(tt.wantTsec) {
101+
t.Fatalf("Timestamp.Unix() = %d, want %d", h.Timestamp.Unix(), tt.wantTsec)
102+
}
103+
})
104+
}
105+
}
106+
107+
func TestHepTimestampInWindow(t *testing.T) {
108+
now := time.Date(2026, 8, 4, 12, 0, 0, 0, time.UTC)
109+
110+
if !hepTimestampInWindow(now, now) {
111+
t.Fatal("now should be in window")
112+
}
113+
if !hepTimestampInWindow(now.Add(23*time.Hour), now) {
114+
t.Fatal("now+23h should be in window")
115+
}
116+
if hepTimestampInWindow(now.Add(25*time.Hour), now) {
117+
t.Fatal("now+25h should be out of window")
118+
}
119+
if hepTimestampInWindow(now.Add(-11*365*24*time.Hour), now) {
120+
t.Fatal("now-11y should be out of window")
121+
}
122+
if !hepTimestampInWindow(now.Add(-9*365*24*time.Hour), now) {
123+
t.Fatal("now-9y should be in window")
124+
}
125+
}

‎src/storage/ducklake/hep_adapter.go‎

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -178,8 +178,11 @@ func (a *MultiTableAdapter) convertHEPToValues(hep *decoder.HEP) (TableKey, []in
178178
// If forcedSIPSubtype is non-empty and hep is SIP (proto 1), that subtype selects the
179179
// DuckLake table and row shape (call / registration / default) instead of inferring from the method.
180180
func (a *MultiTableAdapter) convertHEPToValuesWithSIPSubtype(hep *decoder.HEP, forcedSIPSubtype string) (TableKey, []interface{}) {
181-
// Calculate timestamp as time.Time for DuckDB TIMESTAMP type
182-
ts := time.Unix(int64(hep.Tsec), int64(hep.Tmsec)*1000)
181+
// Use decoder-sanitized Timestamp (Tsec/Tmsec are kept in sync there).
182+
ts := hep.Timestamp
183+
if ts.IsZero() {
184+
ts = time.Unix(int64(hep.Tsec), int64(hep.Tmsec)*1000)
185+
}
183186
date := fastDuckDate(ts)
184187
uid := fastUUID()
185188
nodeID := fastNodeID(hep.NodeID)
@@ -424,8 +427,11 @@ func (a *MultiTableAdapter) GetReader() *MultiTableReader {
424427

425428
// convertHEP converts a HEP packet to a DuckLake record (legacy)
426429
func (a *HEPAdapter) convertHEP(hep *decoder.HEP) HEPRecord {
427-
// Calculate timestamp as time.Time for DuckDB TIMESTAMP type
428-
ts := time.Unix(int64(hep.Tsec), int64(hep.Tmsec)*1000)
430+
// Use decoder-sanitized Timestamp (Tsec/Tmsec are kept in sync there).
431+
ts := hep.Timestamp
432+
if ts.IsZero() {
433+
ts = time.Unix(int64(hep.Tsec), int64(hep.Tmsec)*1000)
434+
}
429435

430436
record := HEPRecord{
431437
UUID: fastUUID(),

‎src/version.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ import (
2424
// Version information for homer-core
2525
var (
2626
// VERSION_APPLICATION is the application version
27-
VERSION_APPLICATION = "11.0.305"
27+
VERSION_APPLICATION = "11.0.306"
2828

2929
// BuildDate is the build date
3030
BuildDate = ""

0 commit comments

Comments
 (0)