Extract arpa reverse-name parsing into per-family helpers

Expose a network-free login-required check backed by the in-memory status
recorder. Unlike IsLoginRequired(), which creates a fresh auth client and
performs a blocking network call, IsLoginRequiredCached() reports whether the
LAST observed management error was an auth failure (PermissionDenied/
InvalidArgument).

This lets the iOS connection listener detect a mid-session token expiry from
within onDisconnected during teardown without blocking on a slow or
unavailable network.

client/internal/dns/resutil/resolve.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"net"
 	"net/netip"
+	"slices"
 	"strings"
 
 	"github.com/miekg/dns"
@@ -167,7 +168,10 @@ func getRcodeForNotFound(ctx context.Context, r resolver, domain string, origina
 	case dns.TypeA:
 		alternativeNetwork = "ip6"
 	default:
-		return dns.RcodeNameError
+		// Non-address types reach LookupIP only unexpectedly; without an
+		// address pair to probe we cannot prove the name is absent, so answer
+		// NODATA rather than a poisoning NXDOMAIN.
+		return dns.RcodeSuccess
 	}
 
 	if _, err := r.LookupNetIP(ctx, alternativeNetwork, domain); err != nil {
@@ -184,6 +188,230 @@ func getRcodeForNotFound(ctx context.Context, r resolver, domain string, origina
 	return dns.RcodeSuccess
 }
 
+// RecordResolver is the host resolver surface used to forward non-address
+// record queries. net.DefaultResolver satisfies it.
+type RecordResolver interface {
+	LookupMX(ctx context.Context, name string) ([]*net.MX, error)
+	LookupTXT(ctx context.Context, name string) ([]string, error)
+	LookupNS(ctx context.Context, name string) ([]*net.NS, error)
+	LookupSRV(ctx context.Context, service, proto, name string) (string, []*net.SRV, error)
+	LookupCNAME(ctx context.Context, host string) (string, error)
+	LookupAddr(ctx context.Context, addr string) ([]string, error)
+}
+
+// LookupRecords resolves a non-address DNS record type through the host
+// resolver and returns the resource records and the DNS rcode. Types the host
+// resolver cannot answer (anything not covered by the net.Resolver Lookup*
+// methods) yield NODATA so that a routed name is never poisoned with NXDOMAIN
+// for an unsupported type.
+func LookupRecords(ctx context.Context, r RecordResolver, name string, qtype uint16, ttl uint32) ([]dns.RR, int) {
+	fqdn := dns.Fqdn(name)
+
+	switch qtype {
+	case dns.TypeMX:
+		return lookupMX(ctx, r, name, fqdn, ttl)
+	case dns.TypeTXT:
+		return lookupTXT(ctx, r, name, fqdn, ttl)
+	case dns.TypeNS:
+		return lookupNS(ctx, r, name, fqdn, ttl)
+	case dns.TypeSRV:
+		return lookupSRV(ctx, r, name, fqdn, ttl)
+	case dns.TypeCNAME:
+		return lookupCNAME(ctx, r, name, fqdn, ttl)
+	case dns.TypePTR:
+		return lookupPTR(ctx, r, name, fqdn, ttl)
+	default:
+		return nil, dns.RcodeSuccess
+	}
+}
+
+func recordHeader(fqdn string, rrtype uint16, ttl uint32) dns.RR_Header {
+	return dns.RR_Header{Name: fqdn, Rrtype: rrtype, Class: dns.ClassINET, Ttl: ttl}
+}
+
+func lookupMX(ctx context.Context, r RecordResolver, name, fqdn string, ttl uint32) ([]dns.RR, int) {
+	recs, err := r.LookupMX(ctx, name)
+	if err != nil {
+		return nil, rcodeForRecordError(err)
+	}
+	rrs := make([]dns.RR, 0, len(recs))
+	for _, mx := range recs {
+		rrs = append(rrs, &dns.MX{
+			Hdr:        recordHeader(fqdn, dns.TypeMX, ttl),
+			Preference: mx.Pref,
+			Mx:         dns.Fqdn(mx.Host),
+		})
+	}
+	return rrs, dns.RcodeSuccess
+}
+
+func lookupTXT(ctx context.Context, r RecordResolver, name, fqdn string, ttl uint32) ([]dns.RR, int) {
+	recs, err := r.LookupTXT(ctx, name)
+	if err != nil {
+		return nil, rcodeForRecordError(err)
+	}
+	rrs := make([]dns.RR, 0, len(recs))
+	for _, txt := range recs {
+		rrs = append(rrs, &dns.TXT{
+			Hdr: recordHeader(fqdn, dns.TypeTXT, ttl),
+			Txt: chunkTXT(txt),
+		})
+	}
+	return rrs, dns.RcodeSuccess
+}
+
+func lookupNS(ctx context.Context, r RecordResolver, name, fqdn string, ttl uint32) ([]dns.RR, int) {
+	recs, err := r.LookupNS(ctx, name)
+	if err != nil {
+		return nil, rcodeForRecordError(err)
+	}
+	rrs := make([]dns.RR, 0, len(recs))
+	for _, ns := range recs {
+		rrs = append(rrs, &dns.NS{
+			Hdr: recordHeader(fqdn, dns.TypeNS, ttl),
+			Ns:  dns.Fqdn(ns.Host),
+		})
+	}
+	return rrs, dns.RcodeSuccess
+}
+
+func lookupSRV(ctx context.Context, r RecordResolver, name, fqdn string, ttl uint32) ([]dns.RR, int) {
+	_, recs, err := r.LookupSRV(ctx, "", "", name)
+	if err != nil {
+		return nil, rcodeForRecordError(err)
+	}
+	rrs := make([]dns.RR, 0, len(recs))
+	for _, srv := range recs {
+		rrs = append(rrs, &dns.SRV{
+			Hdr:      recordHeader(fqdn, dns.TypeSRV, ttl),
+			Priority: srv.Priority,
+			Weight:   srv.Weight,
+			Port:     srv.Port,
+			Target:   dns.Fqdn(srv.Target),
+		})
+	}
+	return rrs, dns.RcodeSuccess
+}
+
+func lookupCNAME(ctx context.Context, r RecordResolver, name, fqdn string, ttl uint32) ([]dns.RR, int) {
+	cname, err := r.LookupCNAME(ctx, name)
+	if err != nil {
+		return nil, rcodeForRecordError(err)
+	}
+	// LookupCNAME returns the queried name itself when the name resolves but
+	// has no CNAME record; that is a NODATA result, not a CNAME.
+	if strings.EqualFold(dns.Fqdn(cname), fqdn) {
+		return nil, dns.RcodeSuccess
+	}
+	return []dns.RR{&dns.CNAME{
+		Hdr:    recordHeader(fqdn, dns.TypeCNAME, ttl),
+		Target: dns.Fqdn(cname),
+	}}, dns.RcodeSuccess
+}
+
+func lookupPTR(ctx context.Context, r RecordResolver, name, fqdn string, ttl uint32) ([]dns.RR, int) {
+	addr, ok := ptrQueryAddr(name)
+	if !ok {
+		return nil, dns.RcodeSuccess
+	}
+	names, err := r.LookupAddr(ctx, addr)
+	if err != nil {
+		return nil, rcodeForRecordError(err)
+	}
+	rrs := make([]dns.RR, 0, len(names))
+	for _, n := range names {
+		rrs = append(rrs, &dns.PTR{
+			Hdr: recordHeader(fqdn, dns.TypePTR, ttl),
+			Ptr: dns.Fqdn(n),
+		})
+	}
+	return rrs, dns.RcodeSuccess
+}
+
+// ptrQueryAddr converts a reverse-DNS query name (in-addr.arpa or ip6.arpa)
+// into the address string expected by net.Resolver.LookupAddr. It reports false
+// when the name is not a well-formed reverse name.
+func ptrQueryAddr(qname string) (string, bool) {
+	name := strings.TrimSuffix(strings.ToLower(dns.Fqdn(qname)), ".")
+
+	switch {
+	case strings.HasSuffix(name, ".in-addr.arpa"):
+		return parseInAddrArpa(strings.TrimSuffix(name, ".in-addr.arpa"))
+	case strings.HasSuffix(name, ".ip6.arpa"):
+		return parseIP6Arpa(strings.TrimSuffix(name, ".ip6.arpa"))
+	default:
+		return "", false
+	}
+}
+
+// parseInAddrArpa turns the label portion of an in-addr.arpa name into an IPv4
+// address string, reporting false when it is not a well-formed reverse name.
+func parseInAddrArpa(labelPart string) (string, bool) {
+	labels := strings.Split(labelPart, ".")
+	if len(labels) != 4 {
+		return "", false
+	}
+	slices.Reverse(labels)
+	addr, err := netip.ParseAddr(strings.Join(labels, "."))
+	if err != nil || !addr.Is4() {
+		return "", false
+	}
+	return addr.String(), true
+}
+
+// parseIP6Arpa turns the nibble portion of an ip6.arpa name into an IPv6
+// address string, reporting false when it is not a well-formed reverse name.
+func parseIP6Arpa(nibblePart string) (string, bool) {
+	nibbles := strings.Split(nibblePart, ".")
+	if len(nibbles) != 32 {
+		return "", false
+	}
+	slices.Reverse(nibbles)
+	var sb strings.Builder
+	for i, n := range nibbles {
+		if i > 0 && i%4 == 0 {
+			sb.WriteByte(':')
+		}
+		sb.WriteString(n)
+	}
+	addr, err := netip.ParseAddr(sb.String())
+	if err != nil || !addr.Is6() {
+		return "", false
+	}
+	return addr.String(), true
+}
+
+// rcodeForRecordError maps a non-address lookup error to a DNS rcode. A
+// not-found result becomes NODATA rather than NXDOMAIN: net.DNSError.IsNotFound
+// does not distinguish a missing name from a name that exists only with records
+// of other types, so the name cannot be proven absent and must not be poisoned.
+func rcodeForRecordError(err error) int {
+	var dnsErr *net.DNSError
+	if errors.As(err, &dnsErr) && dnsErr.IsNotFound {
+		return dns.RcodeSuccess
+	}
+	return dns.RcodeServerFailure
+}
+
+// chunkTXT splits a TXT string into character-strings no longer than 255 bytes
+// so the record can be packed. The chunks form one TXT resource record.
+func chunkTXT(s string) []string {
+	const maxLen = 255
+	if len(s) <= maxLen {
+		return []string{s}
+	}
+
+	var chunks []string
+	for len(s) > maxLen {
+		chunks = append(chunks, s[:maxLen])
+		s = s[maxLen:]
+	}
+	if len(s) > 0 {
+		chunks = append(chunks, s)
+	}
+	return chunks
+}
+
 // FormatAnswers formats DNS resource records for logging.
 func FormatAnswers(answers []dns.RR) string {
 	if len(answers) == 0 {

client/internal/dns/resutil/resolve_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"net"
 	"net/netip"
+	"strings"
 	"testing"
 
 	"github.com/miekg/dns"
@@ -120,3 +121,161 @@ func TestLookupIP_DNSErrorNotIsNotFound(t *testing.T) {
 
 	assert.Equal(t, dns.RcodeServerFailure, result.Rcode, "upstream failure should map to SERVFAIL")
 }
+
+func TestPtrQueryAddr(t *testing.T) {
+	tests := []struct {
+		name   string
+		qname  string
+		want   string
+		wantOK bool
+	}{
+		{name: "ipv4", qname: "4.3.2.1.in-addr.arpa.", want: "1.2.3.4", wantOK: true},
+		{name: "ipv4 no trailing dot", qname: "1.0.0.127.in-addr.arpa", want: "127.0.0.1", wantOK: true},
+		{
+			name:   "ipv6",
+			qname:  "1.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.8.b.d.0.1.0.0.2.ip6.arpa.",
+			want:   "2001:db8::1",
+			wantOK: true,
+		},
+		{name: "ipv4 wrong label count", qname: "2.1.in-addr.arpa.", wantOK: false},
+		{name: "ipv6 wrong nibble count", qname: "1.0.ip6.arpa.", wantOK: false},
+		{name: "not a reverse name", qname: "example.com.", wantOK: false},
+		{name: "ipv4 bad octet", qname: "4.3.2.999.in-addr.arpa.", wantOK: false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, ok := ptrQueryAddr(tt.qname)
+			assert.Equal(t, tt.wantOK, ok, "parse success mismatch")
+			if tt.wantOK {
+				assert.Equal(t, tt.want, got, "parsed address mismatch")
+			}
+		})
+	}
+}
+
+type mockRecordResolver struct {
+	mx    []*net.MX
+	txt   []string
+	ns    []*net.NS
+	srv   []*net.SRV
+	cname string
+	ptr   []string
+	err   error
+}
+
+func (m *mockRecordResolver) LookupMX(context.Context, string) ([]*net.MX, error) {
+	return m.mx, m.err
+}
+func (m *mockRecordResolver) LookupTXT(context.Context, string) ([]string, error) {
+	return m.txt, m.err
+}
+func (m *mockRecordResolver) LookupNS(context.Context, string) ([]*net.NS, error) {
+	return m.ns, m.err
+}
+func (m *mockRecordResolver) LookupSRV(context.Context, string, string, string) (string, []*net.SRV, error) {
+	return "", m.srv, m.err
+}
+func (m *mockRecordResolver) LookupCNAME(context.Context, string) (string, error) {
+	return m.cname, m.err
+}
+func (m *mockRecordResolver) LookupAddr(context.Context, string) ([]string, error) {
+	return m.ptr, m.err
+}
+
+func TestLookupRecords(t *testing.T) {
+	notFound := &net.DNSError{IsNotFound: true, Name: "example.com."}
+
+	t.Run("MX success", func(t *testing.T) {
+		r := &mockRecordResolver{mx: []*net.MX{{Host: "mail.example.com.", Pref: 10}}}
+		rrs, rcode := LookupRecords(context.Background(), r, "example.com.", dns.TypeMX, 300)
+		assert.Equal(t, dns.RcodeSuccess, rcode)
+		require.Len(t, rrs, 1)
+		assert.Equal(t, "mail.example.com.", rrs[0].(*dns.MX).Mx)
+	})
+
+	t.Run("TXT short string is one character-string", func(t *testing.T) {
+		r := &mockRecordResolver{txt: []string{"v=spf1 -all"}}
+		rrs, rcode := LookupRecords(context.Background(), r, "example.com.", dns.TypeTXT, 300)
+		assert.Equal(t, dns.RcodeSuccess, rcode)
+		require.Len(t, rrs, 1)
+		assert.Equal(t, []string{"v=spf1 -all"}, rrs[0].(*dns.TXT).Txt)
+	})
+
+	t.Run("TXT chunks long strings", func(t *testing.T) {
+		long := strings.Repeat("a", 300)
+		r := &mockRecordResolver{txt: []string{long}}
+		rrs, rcode := LookupRecords(context.Background(), r, "example.com.", dns.TypeTXT, 300)
+		assert.Equal(t, dns.RcodeSuccess, rcode)
+		require.Len(t, rrs, 1)
+		txt := rrs[0].(*dns.TXT).Txt
+		require.Len(t, txt, 2, "300-byte string should split into two character-strings")
+		assert.Equal(t, 255, len(txt[0]))
+		assert.Equal(t, 45, len(txt[1]))
+	})
+
+	t.Run("NS success", func(t *testing.T) {
+		r := &mockRecordResolver{ns: []*net.NS{{Host: "ns1.example.com."}}}
+		rrs, rcode := LookupRecords(context.Background(), r, "example.com.", dns.TypeNS, 300)
+		assert.Equal(t, dns.RcodeSuccess, rcode)
+		require.Len(t, rrs, 1)
+		assert.Equal(t, "ns1.example.com.", rrs[0].(*dns.NS).Ns)
+	})
+
+	t.Run("SRV success", func(t *testing.T) {
+		r := &mockRecordResolver{srv: []*net.SRV{{Target: "sip.example.com.", Port: 5060}}}
+		rrs, rcode := LookupRecords(context.Background(), r, "_sip._tcp.example.com.", dns.TypeSRV, 300)
+		assert.Equal(t, dns.RcodeSuccess, rcode)
+		require.Len(t, rrs, 1)
+		assert.Equal(t, uint16(5060), rrs[0].(*dns.SRV).Port)
+	})
+
+	t.Run("CNAME success", func(t *testing.T) {
+		r := &mockRecordResolver{cname: "target.example.com."}
+		rrs, rcode := LookupRecords(context.Background(), r, "www.example.com.", dns.TypeCNAME, 300)
+		assert.Equal(t, dns.RcodeSuccess, rcode)
+		require.Len(t, rrs, 1)
+		assert.Equal(t, "target.example.com.", rrs[0].(*dns.CNAME).Target)
+	})
+
+	t.Run("CNAME equal to name is NODATA", func(t *testing.T) {
+		r := &mockRecordResolver{cname: "example.com."}
+		rrs, rcode := LookupRecords(context.Background(), r, "example.com.", dns.TypeCNAME, 300)
+		assert.Equal(t, dns.RcodeSuccess, rcode)
+		assert.Empty(t, rrs, "self-referential CNAME is NODATA")
+	})
+
+	t.Run("PTR success", func(t *testing.T) {
+		r := &mockRecordResolver{ptr: []string{"host.example.com."}}
+		rrs, rcode := LookupRecords(context.Background(), r, "4.3.2.1.in-addr.arpa.", dns.TypePTR, 300)
+		assert.Equal(t, dns.RcodeSuccess, rcode)
+		require.Len(t, rrs, 1)
+		assert.Equal(t, "host.example.com.", rrs[0].(*dns.PTR).Ptr)
+	})
+
+	t.Run("PTR malformed name is NODATA", func(t *testing.T) {
+		r := &mockRecordResolver{}
+		rrs, rcode := LookupRecords(context.Background(), r, "example.com.", dns.TypePTR, 300)
+		assert.Equal(t, dns.RcodeSuccess, rcode)
+		assert.Empty(t, rrs)
+	})
+
+	t.Run("not found is NODATA never NXDOMAIN", func(t *testing.T) {
+		r := &mockRecordResolver{err: notFound}
+		_, rcode := LookupRecords(context.Background(), r, "example.com.", dns.TypeMX, 300)
+		assert.Equal(t, dns.RcodeSuccess, rcode, "missing record must not poison the name")
+	})
+
+	t.Run("server failure maps to SERVFAIL", func(t *testing.T) {
+		r := &mockRecordResolver{err: &net.DNSError{Err: "server misbehaving", IsTemporary: true}}
+		_, rcode := LookupRecords(context.Background(), r, "example.com.", dns.TypeMX, 300)
+		assert.Equal(t, dns.RcodeServerFailure, rcode)
+	})
+
+	t.Run("unsupported type is NODATA", func(t *testing.T) {
+		r := &mockRecordResolver{}
+		rrs, rcode := LookupRecords(context.Background(), r, "example.com.", dns.TypeCAA, 300)
+		assert.Equal(t, dns.RcodeSuccess, rcode)
+		assert.Empty(t, rrs)
+	})
+}

client/internal/dnsfwd/forwarder.go

@@ -28,6 +28,12 @@ const upstreamTimeout = 15 * time.Second
 
 type resolver interface {
 	LookupNetIP(ctx context.Context, network, host string) ([]netip.Addr, error)
+	LookupMX(ctx context.Context, name string) ([]*net.MX, error)
+	LookupTXT(ctx context.Context, name string) ([]string, error)
+	LookupNS(ctx context.Context, name string) ([]*net.NS, error)
+	LookupSRV(ctx context.Context, service, proto, name string) (string, []*net.SRV, error)
+	LookupCNAME(ctx context.Context, host string) (string, error)
+	LookupAddr(ctx context.Context, addr string) ([]string, error)
 }
 
 type firewaller interface {
@@ -201,12 +207,6 @@ func (f *DNSForwarder) handleDNSQuery(logger *log.Entry, w dns.ResponseWriter, q
 		qname, dns.TypeToString[question.Qtype], dns.ClassToString[question.Qclass])
 
 	resp := query.SetReply(query)
-	network := resutil.NetworkForQtype(question.Qtype)
-	if network == "" {
-		resp.Rcode = dns.RcodeNotImplemented
-		f.writeResponse(logger, w, resp, qname, startTime)
-		return
-	}
 
 	mostSpecificResId, matchingEntries := f.getMatchingEntries(strings.TrimSuffix(qname, "."))
 	if mostSpecificResId == "" {
@@ -218,6 +218,40 @@ func (f *DNSForwarder) handleDNSQuery(logger *log.Entry, w dns.ResponseWriter, q
 	ctx, cancel := context.WithTimeout(context.Background(), upstreamTimeout)
 	defer cancel()
 
+	switch question.Qtype {
+	case dns.TypeA, dns.TypeAAAA:
+		f.handleAddressQuery(ctx, logger, w, resp, mostSpecificResId, matchingEntries, startTime)
+	case dns.TypeMX, dns.TypeTXT, dns.TypeNS, dns.TypeSRV, dns.TypeCNAME, dns.TypePTR:
+		f.handleRecordQuery(ctx, logger, w, resp, startTime)
+	default:
+		// The domain is routed here, so any other type is answered NODATA
+		// (NOERROR, empty answer) rather than falling back to a resolver that
+		// would poison the name with NXDOMAIN. The Extended DNS Error lets a
+		// client tell this capability-driven NODATA apart from an
+		// authoritative one. The OPT pseudo-record must not appear unless the
+		// query advertised EDNS0.
+		if query.IsEdns0() != nil {
+			attachEDE(resp, dns.ExtendedErrorCodeNotSupported, "netbird forwarder: unsupported query type")
+		}
+		f.writeResponse(logger, w, resp, qname, startTime)
+	}
+}
+
+// handleAddressQuery resolves A/AAAA queries, programs the firewall sets and
+// resolved-IP state, and caches the answer for resilience on upstream failure.
+func (f *DNSForwarder) handleAddressQuery(
+	ctx context.Context,
+	logger *log.Entry,
+	w dns.ResponseWriter,
+	resp *dns.Msg,
+	mostSpecificResId route.ResID,
+	matchingEntries []*ForwarderEntry,
+	startTime time.Time,
+) {
+	question := resp.Question[0]
+	qname := strings.ToLower(question.Name)
+
+	network := resutil.NetworkForQtype(question.Qtype)
 	result := resutil.LookupIP(ctx, f.resolver, network, qname, question.Qtype)
 	if result.Err != nil {
 		f.handleDNSError(ctx, logger, w, question, resp, qname, result, startTime)
@@ -231,6 +265,25 @@ func (f *DNSForwarder) handleDNSQuery(logger *log.Entry, w dns.ResponseWriter, q
 	f.writeResponse(logger, w, resp, qname, startTime)
 }
 
+// handleRecordQuery resolves non-address record types (MX, TXT, NS, SRV,
+// CNAME, PTR) through the host resolver. Missing records are answered NODATA so
+// the routed name is never poisoned with NXDOMAIN.
+func (f *DNSForwarder) handleRecordQuery(
+	ctx context.Context,
+	logger *log.Entry,
+	w dns.ResponseWriter,
+	resp *dns.Msg,
+	startTime time.Time,
+) {
+	question := resp.Question[0]
+	qname := strings.ToLower(question.Name)
+
+	records, rcode := resutil.LookupRecords(ctx, f.resolver, qname, question.Qtype, f.ttl)
+	resp.Rcode = rcode
+	resp.Answer = append(resp.Answer, records...)
+	f.writeResponse(logger, w, resp, qname, startTime)
+}
+
 func (f *DNSForwarder) writeResponse(logger *log.Entry, w dns.ResponseWriter, resp *dns.Msg, qname string, startTime time.Time) {
 	if err := w.WriteMsg(resp); err != nil {
 		logger.Errorf("failed to write DNS response: %v", err)
@@ -414,3 +467,14 @@ func (f *DNSForwarder) getMatchingEntries(domain string) (route.ResID, []*Forwar
 
 	return selectedResId, matches
 }
+
+// attachEDE adds an Extended DNS Error (RFC 8914) option to the response,
+// creating the OPT pseudo-record if the response does not already carry one.
+func attachEDE(resp *dns.Msg, code uint16, text string) {
+	opt := resp.IsEdns0()
+	if opt == nil {
+		resp.SetEdns0(dns.DefaultMsgSize, false)
+		opt = resp.IsEdns0()
+	}
+	opt.Option = append(opt.Option, &dns.EDNS0_EDE{InfoCode: code, ExtraText: text})
+}

client/internal/dnsfwd/forwarder_test.go

@@ -132,6 +132,41 @@ func (m *MockResolver) LookupNetIP(ctx context.Context, network, host string) ([
 	return args.Get(0).([]netip.Addr), args.Error(1)
 }
 
+func (m *MockResolver) LookupMX(ctx context.Context, name string) ([]*net.MX, error) {
+	args := m.Called(ctx, name)
+	recs, _ := args.Get(0).([]*net.MX)
+	return recs, args.Error(1)
+}
+
+func (m *MockResolver) LookupTXT(ctx context.Context, name string) ([]string, error) {
+	args := m.Called(ctx, name)
+	recs, _ := args.Get(0).([]string)
+	return recs, args.Error(1)
+}
+
+func (m *MockResolver) LookupNS(ctx context.Context, name string) ([]*net.NS, error) {
+	args := m.Called(ctx, name)
+	recs, _ := args.Get(0).([]*net.NS)
+	return recs, args.Error(1)
+}
+
+func (m *MockResolver) LookupSRV(ctx context.Context, service, proto, name string) (string, []*net.SRV, error) {
+	args := m.Called(ctx, service, proto, name)
+	recs, _ := args.Get(1).([]*net.SRV)
+	return args.String(0), recs, args.Error(2)
+}
+
+func (m *MockResolver) LookupCNAME(ctx context.Context, host string) (string, error) {
+	args := m.Called(ctx, host)
+	return args.String(0), args.Error(1)
+}
+
+func (m *MockResolver) LookupAddr(ctx context.Context, addr string) ([]string, error) {
+	args := m.Called(ctx, addr)
+	recs, _ := args.Get(0).([]string)
+	return recs, args.Error(1)
+}
+
 func TestDNSForwarder_SubdomainAccessLogic(t *testing.T) {
 	tests := []struct {
 		name             string
@@ -544,12 +579,15 @@ func TestDNSForwarder_MultipleIPsInSingleUpdate(t *testing.T) {
 }
 
 func TestDNSForwarder_ResponseCodes(t *testing.T) {
+	// A type with no net.Resolver Lookup method (CAA) must answer NODATA
+	// (NOERROR, empty) rather than NXDOMAIN/NOTIMP to avoid poisoning the name.
 	tests := []struct {
 		name         string
 		queryType    uint16
 		queryDomain  string
 		configured   string
 		expectedCode int
+		expectEDE    bool
 		description  string
 	}{
 		{
@@ -561,28 +599,13 @@ func TestDNSForwarder_ResponseCodes(t *testing.T) {
 			description:  "RFC compliant REFUSED for unauthorized queries",
 		},
 		{
-			name:         "unsupported query type returns NOTIMP",
-			queryType:    dns.TypeMX,
+			name:         "unsupported query type returns NODATA",
+			queryType:    dns.TypeCAA,
 			queryDomain:  "example.com",
 			configured:   "example.com",
-			expectedCode: dns.RcodeNotImplemented,
-			description:  "RFC compliant NOTIMP for unsupported types",
-		},
-		{
-			name:         "CNAME query returns NOTIMP",
-			queryType:    dns.TypeCNAME,
-			queryDomain:  "example.com",
-			configured:   "example.com",
-			expectedCode: dns.RcodeNotImplemented,
-			description:  "CNAME queries not supported",
-		},
-		{
-			name:         "TXT query returns NOTIMP",
-			queryType:    dns.TypeTXT,
-			queryDomain:  "example.com",
-			configured:   "example.com",
-			expectedCode: dns.RcodeNotImplemented,
-			description:  "TXT queries not supported",
+			expectedCode: dns.RcodeSuccess,
+			expectEDE:    true,
+			description:  "Unsupported types answer NODATA, not NXDOMAIN/NOTIMP",
 		},
 	}
 
@@ -598,6 +621,7 @@ func TestDNSForwarder_ResponseCodes(t *testing.T) {
 
 			query := &dns.Msg{}
 			query.SetQuestion(dns.Fqdn(tt.queryDomain), tt.queryType)
+			query.SetEdns0(dns.DefaultMsgSize, false)
 
 			// Capture the written response
 			var writtenResp *dns.Msg
@@ -613,10 +637,213 @@ func TestDNSForwarder_ResponseCodes(t *testing.T) {
 			// Check the response written to the writer
 			require.NotNil(t, writtenResp, "Expected response to be written")
 			assert.Equal(t, tt.expectedCode, writtenResp.Rcode, tt.description)
+			assert.Empty(t, writtenResp.Answer, "Non-address response should carry no answers")
+
+			if tt.expectEDE {
+				require.NotNil(t, writtenResp.IsEdns0(), "EDNS0 client should get an OPT in the reply")
+				assert.True(t, hasEDE(writtenResp, dns.ExtendedErrorCodeNotSupported),
+					"unsupported type NODATA should carry EDE Not Supported")
+			}
 		})
 	}
 }
 
+func hasEDE(m *dns.Msg, code uint16) bool {
+	opt := m.IsEdns0()
+	if opt == nil {
+		return false
+	}
+	for _, o := range opt.Option {
+		if ede, ok := o.(*dns.EDNS0_EDE); ok && ede.InfoCode == code {
+			return true
+		}
+	}
+	return false
+}
+
+func TestDNSForwarder_RecordQueries(t *testing.T) {
+	notFound := &net.DNSError{IsNotFound: true, Name: "example.com"}
+
+	t.Run("MX records are forwarded", func(t *testing.T) {
+		mockResolver := &MockResolver{}
+		forwarder := newRecordTestForwarder(t, mockResolver, "example.com")
+
+		mockResolver.On("LookupMX", mock.Anything, "example.com.").
+			Return([]*net.MX{{Host: "mail.example.com.", Pref: 10}}, nil).Once()
+
+		resp := runRecordQuery(t, forwarder, "example.com", dns.TypeMX)
+		require.Equal(t, dns.RcodeSuccess, resp.Rcode)
+		require.Len(t, resp.Answer, 1)
+		mx, ok := resp.Answer[0].(*dns.MX)
+		require.True(t, ok, "answer should be an MX record")
+		assert.Equal(t, uint16(10), mx.Preference)
+		assert.Equal(t, "mail.example.com.", mx.Mx)
+		mockResolver.AssertExpectations(t)
+	})
+
+	t.Run("missing MX is NODATA not NXDOMAIN", func(t *testing.T) {
+		mockResolver := &MockResolver{}
+		forwarder := newRecordTestForwarder(t, mockResolver, "example.com")
+
+		// A not-found cannot prove the name is absent (it may exist with only
+		// other record types), so it must answer NODATA, never NXDOMAIN.
+		mockResolver.On("LookupMX", mock.Anything, "example.com.").
+			Return(nil, notFound).Once()
+
+		resp := runRecordQuery(t, forwarder, "example.com", dns.TypeMX)
+		assert.Equal(t, dns.RcodeSuccess, resp.Rcode, "missing record must be NODATA")
+		assert.Empty(t, resp.Answer)
+		mockResolver.AssertExpectations(t)
+	})
+
+	t.Run("NS records are forwarded", func(t *testing.T) {
+		mockResolver := &MockResolver{}
+		forwarder := newRecordTestForwarder(t, mockResolver, "example.com")
+
+		mockResolver.On("LookupNS", mock.Anything, "example.com.").
+			Return([]*net.NS{{Host: "ns1.example.com."}}, nil).Once()
+
+		resp := runRecordQuery(t, forwarder, "example.com", dns.TypeNS)
+		require.Equal(t, dns.RcodeSuccess, resp.Rcode)
+		require.Len(t, resp.Answer, 1)
+		ns, ok := resp.Answer[0].(*dns.NS)
+		require.True(t, ok, "answer should be an NS record")
+		assert.Equal(t, "ns1.example.com.", ns.Ns)
+		mockResolver.AssertExpectations(t)
+	})
+
+	t.Run("missing NS is NODATA", func(t *testing.T) {
+		mockResolver := &MockResolver{}
+		forwarder := newRecordTestForwarder(t, mockResolver, "example.com")
+
+		mockResolver.On("LookupNS", mock.Anything, "example.com.").
+			Return(nil, notFound).Once()
+
+		resp := runRecordQuery(t, forwarder, "example.com", dns.TypeNS)
+		assert.Equal(t, dns.RcodeSuccess, resp.Rcode)
+		assert.Empty(t, resp.Answer)
+		mockResolver.AssertExpectations(t)
+	})
+
+	t.Run("SRV records are forwarded", func(t *testing.T) {
+		mockResolver := &MockResolver{}
+		forwarder := newRecordTestForwarder(t, mockResolver, "_sip._tcp.example.com")
+
+		mockResolver.On("LookupSRV", mock.Anything, "", "", "_sip._tcp.example.com.").
+			Return("", []*net.SRV{{Target: "sip.example.com.", Port: 5060, Priority: 10, Weight: 5}}, nil).Once()
+
+		resp := runRecordQuery(t, forwarder, "_sip._tcp.example.com", dns.TypeSRV)
+		require.Equal(t, dns.RcodeSuccess, resp.Rcode)
+		require.Len(t, resp.Answer, 1)
+		srv, ok := resp.Answer[0].(*dns.SRV)
+		require.True(t, ok, "answer should be an SRV record")
+		assert.Equal(t, "sip.example.com.", srv.Target)
+		assert.Equal(t, uint16(5060), srv.Port)
+		assert.Equal(t, uint16(10), srv.Priority)
+		mockResolver.AssertExpectations(t)
+	})
+
+	t.Run("missing SRV is NODATA", func(t *testing.T) {
+		mockResolver := &MockResolver{}
+		forwarder := newRecordTestForwarder(t, mockResolver, "_sip._tcp.example.com")
+
+		mockResolver.On("LookupSRV", mock.Anything, "", "", "_sip._tcp.example.com.").
+			Return("", nil, notFound).Once()
+
+		resp := runRecordQuery(t, forwarder, "_sip._tcp.example.com", dns.TypeSRV)
+		assert.Equal(t, dns.RcodeSuccess, resp.Rcode)
+		assert.Empty(t, resp.Answer)
+		mockResolver.AssertExpectations(t)
+	})
+
+	t.Run("TXT records are forwarded", func(t *testing.T) {
+		mockResolver := &MockResolver{}
+		forwarder := newRecordTestForwarder(t, mockResolver, "example.com")
+
+		mockResolver.On("LookupTXT", mock.Anything, "example.com.").
+			Return([]string{"v=spf1 -all"}, nil).Once()
+
+		resp := runRecordQuery(t, forwarder, "example.com", dns.TypeTXT)
+		require.Equal(t, dns.RcodeSuccess, resp.Rcode)
+		require.Len(t, resp.Answer, 1)
+		txt, ok := resp.Answer[0].(*dns.TXT)
+		require.True(t, ok, "answer should be a TXT record")
+		assert.Equal(t, []string{"v=spf1 -all"}, txt.Txt)
+		mockResolver.AssertExpectations(t)
+	})
+
+	t.Run("CNAME record is forwarded", func(t *testing.T) {
+		mockResolver := &MockResolver{}
+		forwarder := newRecordTestForwarder(t, mockResolver, "www.example.com")
+
+		mockResolver.On("LookupCNAME", mock.Anything, "www.example.com.").
+			Return("target.example.com.", nil).Once()
+
+		resp := runRecordQuery(t, forwarder, "www.example.com", dns.TypeCNAME)
+		require.Equal(t, dns.RcodeSuccess, resp.Rcode)
+		require.Len(t, resp.Answer, 1)
+		cname, ok := resp.Answer[0].(*dns.CNAME)
+		require.True(t, ok, "answer should be a CNAME record")
+		assert.Equal(t, "target.example.com.", cname.Target)
+		mockResolver.AssertExpectations(t)
+	})
+
+	t.Run("CNAME equal to the name is NODATA", func(t *testing.T) {
+		mockResolver := &MockResolver{}
+		forwarder := newRecordTestForwarder(t, mockResolver, "example.com")
+
+		// No CNAME exists: LookupCNAME echoes the queried name back.
+		mockResolver.On("LookupCNAME", mock.Anything, "example.com.").
+			Return("example.com.", nil).Once()
+
+		resp := runRecordQuery(t, forwarder, "example.com", dns.TypeCNAME)
+		assert.Equal(t, dns.RcodeSuccess, resp.Rcode)
+		assert.Empty(t, resp.Answer, "self-referential CNAME means no CNAME record")
+		mockResolver.AssertExpectations(t)
+	})
+
+	t.Run("PTR record is forwarded", func(t *testing.T) {
+		mockResolver := &MockResolver{}
+		forwarder := newRecordTestForwarder(t, mockResolver, "*.in-addr.arpa")
+
+		// The reverse name is parsed back to the address LookupAddr expects.
+		mockResolver.On("LookupAddr", mock.Anything, "1.2.3.4").
+			Return([]string{"host.example.com."}, nil).Once()
+
+		resp := runRecordQuery(t, forwarder, "4.3.2.1.in-addr.arpa", dns.TypePTR)
+		require.Equal(t, dns.RcodeSuccess, resp.Rcode)
+		require.Len(t, resp.Answer, 1)
+		ptr, ok := resp.Answer[0].(*dns.PTR)
+		require.True(t, ok, "answer should be a PTR record")
+		assert.Equal(t, "host.example.com.", ptr.Ptr)
+		mockResolver.AssertExpectations(t)
+	})
+}
+
+func newRecordTestForwarder(t *testing.T, r resolver, configured string) *DNSForwarder {
+	t.Helper()
+	forwarder := NewDNSForwarder(netip.MustParseAddrPort("127.0.0.1:0"), 300, nil, &peer.Status{}, nil)
+	forwarder.resolver = r
+
+	d, err := domain.FromString(configured)
+	require.NoError(t, err)
+	forwarder.UpdateDomains([]*ForwarderEntry{{Domain: d, ResID: "test-res"}})
+	return forwarder
+}
+
+func runRecordQuery(t *testing.T, forwarder *DNSForwarder, qname string, qtype uint16) *dns.Msg {
+	t.Helper()
+	query := &dns.Msg{}
+	query.SetQuestion(dns.Fqdn(qname), qtype)
+
+	mockWriter := &test.MockResponseWriter{}
+	forwarder.handleDNSQuery(log.NewEntry(log.StandardLogger()), mockWriter, query, time.Now())
+
+	resp := mockWriter.GetLastResponse()
+	require.NotNil(t, resp, "expected response to be written")
+	return resp
+}
+
 func TestDNSForwarder_TCPTruncation(t *testing.T) {
 	// Test that large UDP responses are truncated with TC bit set
 	mockResolver := &MockResolver{}

client/internal/netflow/logger/logger.go

@@ -27,7 +27,7 @@ type Logger struct {
 	wgIfaceNetV6       netip.Prefix
 	dnsCollection      atomic.Bool
 	exitNodeCollection atomic.Bool
-	Store              types.AggregatingStore
+	Store              types.Store
 }
 
 func New(statusRecorder *peer.Status, wgIfaceIPNet, wgIfaceIPNetV6 netip.Prefix) *Logger {
@@ -35,7 +35,7 @@ func New(statusRecorder *peer.Status, wgIfaceIPNet, wgIfaceIPNetV6 netip.Prefix)
 		statusRecorder: statusRecorder,
 		wgIfaceNet:     wgIfaceIPNet,
 		wgIfaceNetV6:   wgIfaceIPNetV6,
-		Store:          store.NewAggregatingMemoryStore(),
+		Store:          store.NewMemoryStore(),
 	}
 }
 
@@ -125,10 +125,6 @@ func (l *Logger) stop() {
 	l.mux.Unlock()
 }
 
-func (l *Logger) ResetAggregationWindow() types.FlowEventAggregator {
-	return l.Store.ResetAggregationWindow()
-}
-
 func (l *Logger) GetEvents() []*types.Event {
 	return l.Store.GetEvents()
 }

client/internal/netflow/manager.go

@@ -9,14 +9,12 @@ import (
 	"sync"
 	"time"
 
-	"github.com/cenkalti/backoff/v4"
 	"github.com/google/uuid"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/netbirdio/netbird/client/internal/netflow/conntrack"
 	"github.com/netbirdio/netbird/client/internal/netflow/logger"
-	"github.com/netbirdio/netbird/client/internal/netflow/store"
 	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/flow/client"
@@ -25,16 +23,14 @@ import (
 
 // Manager handles netflow tracking and logging
 type Manager struct {
-	mux               sync.Mutex
-	shutdownWg        sync.WaitGroup
-	logger            nftypes.FlowLogger
-	flowConfig        *nftypes.FlowConfig
-	conntrack         nftypes.ConnTracker
-	receiverClient    *client.GRPCClient
-	eventsWithoutAcks nftypes.Store
-	publicKey         []byte
-	cancel            context.CancelFunc
-	retryInterval     time.Duration
+	mux            sync.Mutex
+	shutdownWg     sync.WaitGroup
+	logger         nftypes.FlowLogger
+	flowConfig     *nftypes.FlowConfig
+	conntrack      nftypes.ConnTracker
+	receiverClient *client.GRPCClient
+	publicKey      []byte
+	cancel         context.CancelFunc
 }
 
 // NewManager creates a new netflow manager
@@ -52,11 +48,9 @@ func NewManager(iface nftypes.IFaceMapper, publicKey []byte, statusRecorder *pee
 	}
 
 	return &Manager{
-		logger:            flowLogger,
-		conntrack:         ct,
-		publicKey:         publicKey,
-		retryInterval:     time.Second,
-		eventsWithoutAcks: store.NewMemoryStore(),
+		logger:    flowLogger,
+		conntrack: ct,
+		publicKey: publicKey,
 	}
 }
 
@@ -113,7 +107,7 @@ func (m *Manager) resetClient() error {
 	ctx, cancel := context.WithCancel(context.Background())
 	m.cancel = cancel
 
-	m.shutdownWg.Add(3)
+	m.shutdownWg.Add(2)
 	go func() {
 		defer m.shutdownWg.Done()
 		m.receiveACKs(ctx, flowClient)
@@ -122,10 +116,6 @@ func (m *Manager) resetClient() error {
 		defer m.shutdownWg.Done()
 		m.startSender(ctx)
 	}()
-	go func() {
-		defer m.shutdownWg.Done()
-		m.startRetries(ctx)
-	}()
 
 	return nil
 }
@@ -217,15 +207,13 @@ func (m *Manager) startSender(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			collectedEvents := m.logger.ResetAggregationWindow()
-			events := collectedEvents.GetAggregatedEvents()
+			events := m.logger.GetEvents()
 			for _, event := range events {
 				if err := m.send(event); err != nil {
 					log.Errorf("failed to send flow event to server: %v", err)
-				} else {
-					log.Tracef("sent flow event: %s", event.ID)
+					continue
 				}
-				m.eventsWithoutAcks.StoreEvent(event)
+				log.Tracef("sent flow event: %s", event.ID)
 			}
 		}
 	}
@@ -239,7 +227,7 @@ func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient) {
 			return nil
 		}
 		log.Tracef("received flow event ack: %s", id)
-		m.eventsWithoutAcks.DeleteEvents([]uuid.UUID{id})
+		m.logger.DeleteEvents([]uuid.UUID{id})
 		return nil
 	})
 
@@ -248,43 +236,6 @@ func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient) {
 	}
 }
 
-// We effectively never drop events (see MaxInterval), which makes eventsWithoutAcks unbounded.
-// We may want to limit the max size of the store, and start dropping oldest events when the threshold is reached.
-func (m *Manager) startRetries(ctx context.Context) {
-	timer := time.NewTimer(m.retryInterval)
-	retryBackoff := backoff.WithContext(&backoff.ExponentialBackOff{
-		InitialInterval:     1 * time.Second,
-		RandomizationFactor: 0.5,
-		Multiplier:          1.7,
-		MaxInterval:         m.flowConfig.Interval / 2,
-		MaxElapsedTime:      3 * 30 * 24 * time.Hour, // 3 months
-		Stop:                backoff.Stop,
-		Clock:               backoff.SystemClock,
-	}, ctx)
-	defer timer.Stop()
-
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-timer.C:
-			for _, e := range m.eventsWithoutAcks.GetEvents() {
-				if e.Timestamp.Add(time.Second).After(time.Now()) {
-					// grace period on retries to avoid early retries
-					// do not retry if the event is less than 1 sec old
-					continue
-				}
-				if err := m.send(e); err != nil {
-					timer = time.NewTimer(retryBackoff.NextBackOff()) //nolint:staticcheck,wastedassign
-					break
-				}
-			}
-			retryBackoff.Reset()
-			timer = time.NewTimer(time.Second)
-		}
-	}
-}
-
 func (m *Manager) send(event *nftypes.Event) error {
 	m.mux.Lock()
 	client := m.receiverClient
@@ -299,11 +250,9 @@ func (m *Manager) send(event *nftypes.Event) error {
 
 func toProtoEvent(publicKey []byte, event *nftypes.Event) *proto.FlowEvent {
 	protoEvent := &proto.FlowEvent{
-		EventId:     event.ID[:],
-		Timestamp:   timestamppb.New(event.Timestamp),
-		PublicKey:   publicKey,
-		WindowStart: timestamppb.New(event.WindowStart),
-		WindowEnd:   timestamppb.New(event.WindowEnd),
+		EventId:   event.ID[:],
+		Timestamp: timestamppb.New(event.Timestamp),
+		PublicKey: publicKey,
 		FlowFields: &proto.FlowFields{
 			FlowId:           event.FlowID[:],
 			RuleId:           event.RuleID,
@@ -318,9 +267,6 @@ func toProtoEvent(publicKey []byte, event *nftypes.Event) *proto.FlowEvent {
 			TxBytes:          event.TxBytes,
 			SourceResourceId: event.SourceResourceID,
 			DestResourceId:   event.DestResourceID,
-			NumOfStarts:      event.NumOfStarts,
-			NumOfEnds:        event.NumOfEnds,
-			NumOfDrops:       event.NumOfDrops,
 		},
 	}
 

client/internal/netflow/manager_integration_test.go

@@ -1,291 +0,0 @@
-package netflow
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"net"
-	"net/netip"
-	"slices"
-	"testing"
-	"time"
-
-	"github.com/google/uuid"
-	"github.com/netbirdio/netbird/client/iface/wgaddr"
-	"github.com/netbirdio/netbird/client/internal/netflow/types"
-	"github.com/netbirdio/netbird/flow/proto"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
-	"google.golang.org/grpc"
-)
-
-type testServer struct {
-	proto.UnimplementedFlowServiceServer
-	events         chan *proto.FlowEvent
-	acks           chan *proto.FlowEventAck
-	grpcSrv        *grpc.Server
-	addr           string
-	handlerDone    chan struct{} // signaled each time Events() exits
-	handlerStarted chan struct{} // signaled each time Events() begins
-}
-
-func newTestServer(t *testing.T) *testServer {
-	listener, err := net.Listen("tcp", "127.0.0.1:0")
-	require.NoError(t, err)
-
-	s := &testServer{
-		events:         make(chan *proto.FlowEvent, 100),
-		acks:           make(chan *proto.FlowEventAck, 100),
-		grpcSrv:        grpc.NewServer(),
-		addr:           listener.Addr().String(),
-		handlerDone:    make(chan struct{}, 10),
-		handlerStarted: make(chan struct{}, 10),
-	}
-
-	proto.RegisterFlowServiceServer(s.grpcSrv, s)
-
-	go func() {
-		if err := s.grpcSrv.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) {
-			t.Logf("server error: %v", err)
-		}
-	}()
-
-	t.Cleanup(func() {
-		s.grpcSrv.Stop()
-	})
-
-	return s
-}
-
-func (s *testServer) Events(stream proto.FlowService_EventsServer) error {
-	defer func() {
-		select {
-		case s.handlerDone <- struct{}{}:
-		default:
-		}
-	}()
-
-	err := stream.Send(&proto.FlowEventAck{IsInitiator: true})
-	if err != nil {
-		return err
-	}
-
-	select {
-	case s.handlerStarted <- struct{}{}:
-	default:
-	}
-
-	ctx, cancel := context.WithCancel(stream.Context())
-	defer cancel()
-
-	go func() {
-		defer cancel()
-		for {
-			event, err := stream.Recv()
-			if err != nil {
-				return
-			}
-
-			if !event.IsInitiator {
-				select {
-				case s.events <- event:
-				case <-ctx.Done():
-					return
-				}
-			}
-		}
-	}()
-
-	for {
-		select {
-		case ack := <-s.acks:
-			if err := stream.Send(ack); err != nil {
-				return err
-			}
-		case <-ctx.Done():
-			return ctx.Err()
-		}
-	}
-}
-
-func TestSendEventReceiveAck(t *testing.T) {
-	_, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	t.Cleanup(cancel)
-
-	server := newTestServer(t)
-	manager := createManager(t, server.addr, 60*time.Second) // set high to prevent retries in this test
-	defer manager.Close()
-
-	assert.Eventually(t, func() bool {
-		select {
-		case <-server.handlerStarted:
-			return true
-		default:
-			return false
-		}
-	}, 3*time.Second, 100*time.Millisecond)
-
-	event1 := types.EventFields{
-		FlowID:    uuid.New(),
-		Type:      types.TypeStart,
-		Direction: types.Ingress,
-		DestIP:    ipAddr("172.16.1.2"),
-		DestPort:  2345,
-		Protocol:  6,
-	}
-	manager.logger.StoreEvent(event1)
-	event2 := types.EventFields{
-		FlowID:    uuid.New(),
-		Type:      types.TypeStart,
-		Direction: types.Ingress,
-		DestIP:    ipAddr("172.16.1.1"),
-		DestPort:  1234,
-		Protocol:  6,
-	}
-	manager.logger.StoreEvent(event2)
-
-	// verify the server received logged events
-	serverSideEvents := make([]*proto.FlowEvent, 0)
-	assert.Eventually(t, func() bool {
-		select {
-		case event := <-server.events:
-			serverSideEvents = append(serverSideEvents, event)
-			if len(serverSideEvents) == 2 {
-				return true
-			}
-		default:
-			if len(serverSideEvents) == 2 {
-				return true
-			}
-		}
-		return false
-	}, 5*time.Second, 100*time.Millisecond)
-
-	serverSideFlowIds := make([]uuid.UUID, 0, 2)
-	slices.Values(serverSideEvents)(func(e *proto.FlowEvent) bool {
-		id, err := uuid.FromBytes(e.FlowFields.FlowId)
-		assert.NoError(t, err)
-		serverSideFlowIds = append(serverSideFlowIds, id)
-		return true
-	})
-	assert.ElementsMatch(t, []uuid.UUID{event1.FlowID, event2.FlowID}, serverSideFlowIds)
-
-	// verify the manager tracks un-acked events
-	unackedEvents := manager.eventsWithoutAcks.GetEvents()
-	assert.Len(t, unackedEvents, 2)
-	flowIds := make([]uuid.UUID, 0)
-	slices.Values(unackedEvents)(func(e *types.Event) bool {
-		flowIds = append(flowIds, e.FlowID)
-		return true
-	})
-	assert.ElementsMatch(t, flowIds, []uuid.UUID{event1.FlowID, event2.FlowID})
-}
-
-// verify handling of retries:
-//   - unacked events are retried
-//   - when acks arrive, events are removed from the un-acked event tracker
-func TestRetryEvents(t *testing.T) {
-	_, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	t.Cleanup(cancel)
-
-	server := newTestServer(t)
-	manager := createManager(t, server.addr, time.Second) // set low to start retries sooner
-	defer manager.Close()
-
-	assert.Eventually(t, func() bool {
-		select {
-		case <-server.handlerStarted:
-			return true
-		default:
-			return false
-		}
-	}, 3*time.Second, 100*time.Millisecond)
-
-	event1 := types.EventFields{
-		FlowID:    uuid.New(),
-		Type:      types.TypeStart,
-		Direction: types.Ingress,
-		DestIP:    ipAddr("172.16.1.2"),
-		DestPort:  2345,
-		Protocol:  6,
-	}
-	manager.logger.StoreEvent(event1)
-	event2 := types.EventFields{
-		FlowID:    uuid.New(),
-		Type:      types.TypeStart,
-		Direction: types.Ingress,
-		DestIP:    ipAddr("172.16.1.1"),
-		DestPort:  1234,
-		Protocol:  6,
-	}
-	manager.logger.StoreEvent(event2)
-
-	// verify the server received retries of logged events
-	serverSideEvents := make([]*proto.FlowEvent, 0)
-	func() {
-		c := time.After(2500 * time.Millisecond)
-		for {
-			select {
-			case event := <-server.events:
-				serverSideEvents = append(serverSideEvents, event)
-			case <-c:
-				return
-			}
-		}
-	}()
-	assert.True(t, len(serverSideEvents) > 2) // must see retries
-
-	uniqueServerSideEvents := make(map[uuid.UUID]*proto.FlowEvent)
-	slices.Values(serverSideEvents)(func(e *proto.FlowEvent) bool {
-		id, err := uuid.FromBytes(e.FlowFields.FlowId)
-		assert.NoError(t, err)
-		uniqueServerSideEvents[id] = e
-		return true
-	})
-	assert.Contains(t, uniqueServerSideEvents, event1.FlowID)
-	assert.Contains(t, uniqueServerSideEvents, event2.FlowID)
-
-	// ack events
-	server.acks <- &proto.FlowEventAck{EventId: uniqueServerSideEvents[event1.FlowID].EventId}
-	server.acks <- &proto.FlowEventAck{EventId: uniqueServerSideEvents[event2.FlowID].EventId}
-
-	assert.EventuallyWithT(t, func(c *assert.CollectT) {
-		unackedEvents := manager.eventsWithoutAcks.GetEvents()
-		assert.Empty(c, unackedEvents)
-
-	}, 3*time.Second, 100*time.Millisecond)
-}
-
-func createManager(t *testing.T, serverAddr string, retryInterval time.Duration) *Manager {
-	t.Helper()
-
-	mockIFace := &mockIFaceMapper{
-		address: wgaddr.Address{
-			Network: netip.MustParsePrefix("192.168.1.1/32"),
-		},
-		isUserspaceBind: true,
-	}
-
-	publicKey := []byte("test-public-key")
-	manager := NewManager(mockIFace, publicKey, nil)
-	manager.retryInterval = retryInterval
-
-	initialConfig := &types.FlowConfig{
-		Enabled:        true,
-		URL:            fmt.Sprintf("http://%s", serverAddr),
-		TokenPayload:   "initial-payload",
-		TokenSignature: "initial-signature",
-		Interval:       500 * time.Millisecond,
-	}
-
-	err := manager.Update(initialConfig)
-	require.NoError(t, err)
-
-	return manager
-}
-
-func ipAddr(a string) netip.Addr {
-	addr, _ := netip.ParseAddr(a)
-	return addr
-}

client/internal/netflow/store/event_aggregation_test.go

@@ -1,209 +0,0 @@
-package store
-
-import (
-	"math/rand"
-	"net/netip"
-	"testing"
-	"time"
-
-	"github.com/google/uuid"
-	"github.com/netbirdio/netbird/client/internal/netflow/types"
-	"github.com/stretchr/testify/assert"
-)
-
-var random = rand.New(rand.NewSource(time.Now().UnixNano()))
-
-func TestFlowAggregation(t *testing.T) {
-	var protocols = []types.Protocol{types.ICMP, types.ICMPv6, types.TCP, types.UDP}
-	var tests = []struct {
-		description string
-		addresses   [][]netip.Addr
-		dstPort     uint16
-		eventTypes  []types.Type
-	}{
-		{
-			description: "start and stop",
-			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}, {ipAddr("3.3.3.3"), ipAddr("2.2.2.2")}},
-			dstPort:     uint16(random.Uint32() >> 16),
-			eventTypes:  []types.Type{types.TypeStart, types.TypeEnd},
-		},
-		{
-			description: "start and drop",
-			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}, {ipAddr("3.3.3.3"), ipAddr("2.2.2.2")}},
-			dstPort:     uint16(random.Uint32() >> 16),
-			eventTypes:  []types.Type{types.TypeStart, types.TypeDrop},
-		},
-		{
-			description: "start only",
-			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}, {ipAddr("3.3.3.3"), ipAddr("2.2.2.2")}},
-			dstPort:     uint16(random.Uint32() >> 16),
-			eventTypes:  []types.Type{types.TypeStart},
-		},
-		{
-			description: "drop only",
-			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}, {ipAddr("3.3.3.3"), ipAddr("2.2.2.2")}},
-			dstPort:     uint16(random.Uint32() >> 16),
-			eventTypes:  []types.Type{types.TypeDrop},
-		}}
-
-	for _, protocol := range protocols {
-		for _, tt := range tests {
-			t.Run(tt.description+" "+protocol.String(), func(t *testing.T) {
-				store := NewAggregatingMemoryStore()
-				store.WindowEnd = time.Now().Add(5 * time.Second)
-
-				allExpected := make([]*types.Event, 0)
-
-				for _, srcAndDst := range tt.addresses {
-					inEvents, expected := generateEvents(srcAndDst[0], srcAndDst[1], tt.dstPort, tt.eventTypes, protocol, types.Ingress, 0, store.WindowStart, store.WindowEnd)
-					for _, e := range inEvents {
-						store.StoreEvent(e)
-					}
-					allExpected = append(allExpected, expected)
-				}
-
-				events := store.GetAggregatedEvents()
-				assert.ElementsMatch(t, events, allExpected)
-			})
-		}
-	}
-}
-
-func TestIcmpEventAggregation(t *testing.T) {
-	var protocols = []types.Protocol{types.ICMP, types.ICMPv6}
-	var icmpTypes = []uint8{1, 2, 3}
-
-	var tests = []struct {
-		description string
-		addresses   [][]netip.Addr
-		eventTypes  []types.Type
-	}{
-		{
-			description: "start and stop",
-			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}},
-			eventTypes:  []types.Type{types.TypeStart, types.TypeEnd},
-		},
-		{
-			description: "start and drop",
-			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}},
-			eventTypes:  []types.Type{types.TypeStart, types.TypeDrop},
-		},
-		{
-			description: "start only",
-			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}},
-			eventTypes:  []types.Type{types.TypeStart},
-		},
-		{
-			description: "drop only",
-			addresses:   [][]netip.Addr{{ipAddr("1.1.1.1"), ipAddr("2.2.2.2")}},
-			eventTypes:  []types.Type{types.TypeDrop},
-		}}
-
-	for _, protocol := range protocols {
-		for _, tt := range tests {
-			t.Run(tt.description+" "+protocol.String(), func(t *testing.T) {
-				store := NewAggregatingMemoryStore()
-				store.WindowEnd = time.Now().Add(5 * time.Second)
-
-				allExpected := make([]*types.Event, 0)
-				for _, icmpType := range icmpTypes {
-					events, expected := generateEvents(tt.addresses[0][0], tt.addresses[0][1], 0, tt.eventTypes, protocol, types.Ingress, icmpType, store.WindowStart, store.WindowEnd)
-					for _, e := range events {
-						store.StoreEvent(e)
-					}
-					allExpected = append(allExpected, expected)
-				}
-				aggregatedEvents := store.GetAggregatedEvents()
-				assert.Len(t, aggregatedEvents, len(allExpected))
-				assert.ElementsMatch(t, aggregatedEvents, allExpected)
-			})
-		}
-	}
-}
-
-func ipAddr(a string) netip.Addr {
-	addr, _ := netip.ParseAddr(a)
-	return addr
-}
-
-func generateEvents(srcIp, dstIp netip.Addr, dstPort uint16, eventTypes []types.Type, protocol types.Protocol,
-	direction types.Direction, icmpType uint8, windowStart, windowEnd time.Time) ([]*types.Event, *types.Event) {
-	var rxPackets, txPackets, rxBytes, txBytes uint64
-	inEvents := make([]*types.Event, 0)
-	ts := time.Now()
-	flowId := uuid.New()
-	srcPort := uint16(random.Uint32() >> 16)
-
-	for idx, eventType := range eventTypes {
-		e := &types.Event{
-			ID:        uuid.New(),
-			Timestamp: ts.Add(time.Duration(idx) * time.Second),
-			EventFields: types.EventFields{
-				FlowID:           flowId,
-				Type:             eventType,
-				Protocol:         protocol,
-				RuleID:           []byte("rule-id-1"),
-				Direction:        direction,
-				SourceIP:         srcIp,
-				SourcePort:       srcPort,
-				DestIP:           dstIp,
-				DestPort:         dstPort,
-				SourceResourceID: []byte("source-resource-id"),
-				DestResourceID:   []byte("dest-resource-id"),
-				RxPackets:        random.Uint64(),
-				TxPackets:        random.Uint64(),
-				RxBytes:          random.Uint64(),
-				TxBytes:          random.Uint64(),
-			}}
-		rxBytes += e.RxBytes
-		txBytes += e.TxBytes
-		rxPackets += e.RxPackets
-		txPackets += e.TxPackets
-		inEvents = append(inEvents, e)
-		if protocol == types.ICMP || protocol == types.ICMPv6 {
-			e.ICMPType = icmpType
-		}
-	}
-
-	var start, end, drop uint64
-	for _, eventType := range eventTypes {
-		switch eventType {
-		case types.TypeStart:
-			start += 1
-		case types.TypeDrop:
-			drop += 1
-		case types.TypeEnd:
-			end += 1
-		}
-	}
-	aggregatedEvent := &types.Event{
-		ID:          inEvents[0].ID,
-		Timestamp:   inEvents[0].Timestamp,
-		WindowStart: windowStart,
-		WindowEnd:   windowEnd,
-		EventFields: types.EventFields{
-			FlowID:           flowId,
-			Type:             types.TypeUnknown,
-			Protocol:         inEvents[0].Protocol,
-			RuleID:           []byte("rule-id-1"),
-			Direction:        inEvents[0].Direction,
-			SourceIP:         srcIp,
-			SourcePort:       srcPort,
-			DestIP:           dstIp,
-			DestPort:         dstPort,
-			SourceResourceID: []byte("source-resource-id"),
-			DestResourceID:   []byte("dest-resource-id"),
-			RxPackets:        rxPackets,
-			TxPackets:        txPackets,
-			RxBytes:          rxBytes,
-			TxBytes:          txBytes,
-			NumOfStarts:      start,
-			NumOfEnds:        end,
-			NumOfDrops:       drop,
-		}}
-	if protocol == types.ICMP || protocol == types.ICMPv6 {
-		aggregatedEvent.ICMPType = icmpType
-	}
-
-	return inEvents, aggregatedEvent
-}

client/internal/netflow/store/memory.go

@@ -1,13 +1,10 @@
 package store
 
 import (
-	"maps"
-	"net/netip"
-	"slices"
 	"sync"
-	"time"
 
 	"github.com/google/uuid"
+
 	"github.com/netbirdio/netbird/client/internal/netflow/types"
 )
 
@@ -22,12 +19,6 @@ type Memory struct {
 	events map[uuid.UUID]*types.Event
 }
 
-type AggregatingMemory struct {
-	Memory
-	WindowStart time.Time
-	WindowEnd   time.Time
-}
-
 func (m *Memory) StoreEvent(event *types.Event) {
 	m.mux.Lock()
 	defer m.mux.Unlock()
@@ -57,92 +48,3 @@ func (m *Memory) DeleteEvents(ids []uuid.UUID) {
 		delete(m.events, id)
 	}
 }
-
-func NewAggregatingMemoryStore() *AggregatingMemory {
-	return &AggregatingMemory{WindowStart: time.Now(), Memory: Memory{events: make(map[uuid.UUID]*types.Event)}}
-}
-
-func (am *AggregatingMemory) ResetAggregationWindow() types.FlowEventAggregator {
-	am.mux.Lock()
-	defer am.mux.Unlock()
-
-	toret := AggregatingMemory{WindowStart: am.WindowStart, WindowEnd: time.Now(), Memory: Memory{events: am.events}}
-
-	am.events = make(map[uuid.UUID]*types.Event)
-	am.WindowStart = time.Now()
-
-	return &toret
-}
-
-type aggregationKey struct {
-	srcAddr   netip.Addr
-	destAddr  netip.Addr
-	destPort  uint16
-	direction int
-	protocol  uint8
-	icmpType  uint8
-	unique    int64 // used to prevent aggregation on non icmp/udp/tcp events
-}
-
-func (am *AggregatingMemory) GetAggregatedEvents() []*types.Event {
-	am.mux.Lock()
-	defer am.mux.Unlock()
-
-	aggregated := make(map[aggregationKey]*types.Event)
-	for _, v := range am.events {
-		lookupKey := aggregationKey{srcAddr: v.SourceIP, destAddr: v.DestIP, destPort: v.DestPort, direction: int(v.Direction), protocol: uint8(v.Protocol), icmpType: v.ICMPType}
-		if _, ok := aggregated[lookupKey]; !ok {
-			event := v.Clone()
-
-			if event.Protocol != types.ICMP && event.Protocol != types.ICMPv6 && event.Protocol != types.UDP && event.Protocol != types.TCP {
-				lookupKey.unique = time.Now().UnixNano() // to make the lookup key unique so we don't aggregate on it
-				aggregated[lookupKey] = event
-				continue
-			}
-			aggregated[lookupKey] = event
-
-			switch event.Type {
-			case types.TypeStart:
-				event.NumOfStarts += 1
-			case types.TypeDrop:
-				event.NumOfDrops += 1
-			case types.TypeEnd:
-				event.NumOfEnds += 1
-			}
-			event.Type = types.TypeUnknown
-
-			// Please note that ICMPCode field isn't propagated by the manager (see flow/proto/flow.pb.go, FlowFields struct)
-			// so the field value in an icmp event in the "aggregated" doesn't matter
-
-			event.WindowStart = am.WindowStart
-			event.WindowEnd = am.WindowEnd
-			continue
-		}
-
-		aggregatedEvent := aggregated[lookupKey]
-		if aggregatedEvent.Protocol != types.ICMP && aggregatedEvent.Protocol != types.ICMPv6 && aggregatedEvent.Protocol != types.UDP && aggregatedEvent.Protocol != types.TCP {
-			continue // we don't aggregate this type of events; shouldn't ever get here
-		}
-
-		// track the number of connections, duration?, open and close events?
-		aggregatedEvent.RxBytes += v.RxBytes
-		aggregatedEvent.RxPackets += v.RxPackets
-		aggregatedEvent.TxBytes += v.TxBytes
-		aggregatedEvent.TxPackets += v.TxPackets
-		switch v.Type {
-		case types.TypeStart:
-			aggregatedEvent.NumOfStarts += 1
-		case types.TypeDrop:
-			aggregatedEvent.NumOfDrops += 1
-		case types.TypeEnd:
-			aggregatedEvent.NumOfEnds += 1
-		}
-		if aggregatedEvent.Timestamp.Compare(v.Timestamp) > 0 {
-			aggregatedEvent.Timestamp = v.Timestamp
-			aggregatedEvent.ID = v.ID
-			aggregatedEvent.SourcePort = v.SourcePort
-		}
-	}
-
-	return slices.Collect(maps.Values(aggregated)) // could return an iterator instead here
-}

client/internal/netflow/types/types.go

@@ -2,7 +2,6 @@ package types
 
 import (
 	"net/netip"
-	"slices"
 	"strconv"
 	"time"
 
@@ -70,10 +69,8 @@ const (
 )
 
 type Event struct {
-	ID          uuid.UUID
-	Timestamp   time.Time
-	WindowStart time.Time
-	WindowEnd   time.Time
+	ID        uuid.UUID
+	Timestamp time.Time
 	EventFields
 }
 
@@ -95,17 +92,6 @@ type EventFields struct {
 	TxPackets        uint64
 	RxBytes          uint64
 	TxBytes          uint64
-	NumOfStarts      uint64
-	NumOfEnds        uint64
-	NumOfDrops       uint64
-}
-
-func (e *Event) Clone() *Event {
-	toret := *e
-	toret.RuleID = slices.Clone(e.RuleID)
-	toret.SourceResourceID = slices.Clone(e.SourceResourceID)
-	toret.DestResourceID = slices.Clone(e.DestResourceID)
-	return &toret
 }
 
 type FlowConfig struct {
@@ -128,15 +114,13 @@ type FlowManager interface {
 	GetLogger() FlowLogger
 }
 
-type FlowEventAggregator interface {
-	ResetAggregationWindow() FlowEventAggregator
-	GetAggregatedEvents() []*Event
-}
-
 type FlowLogger interface {
-	ResetAggregationWindow() FlowEventAggregator
 	// StoreEvent stores a flow event
 	StoreEvent(flowEvent EventFields)
+	// GetEvents returns all stored events
+	GetEvents() []*Event
+	// DeleteEvents deletes events from the store
+	DeleteEvents([]uuid.UUID)
 	// Close closes the logger
 	Close()
 	// Enable enables the flow logger receiver
@@ -156,11 +140,6 @@ type Store interface {
 	Close()
 }
 
-type AggregatingStore interface {
-	FlowEventAggregator
-	Store
-}
-
 // ConnTracker defines the interface for connection tracking functionality
 type ConnTracker interface {
 	// Start begins tracking connections by listening for conntrack events.

client/internal/routemanager/dnsinterceptor/handler.go

@@ -226,12 +226,11 @@ func (d *DnsInterceptor) ServeDNS(w dns.ResponseWriter, r *dns.Msg) {
 		return
 	}
 
-	// pass if non A/AAAA query
-	if r.Question[0].Qtype != dns.TypeA && r.Question[0].Qtype != dns.TypeAAAA {
-		d.continueToNextHandler(w, r, logger, "non A/AAAA query")
-		return
-	}
-
+	// All query types for an intercepted domain are forwarded to the peer's
+	// DNS forwarder, which owns the name. Falling through to the system
+	// resolver would let it answer NXDOMAIN for a name it isn't authoritative
+	// for, poisoning the whole name (including the A/AAAA records the route
+	// does serve). The forwarder answers NODATA for types it cannot resolve.
 	d.mu.RLock()
 	peerKey := d.currentPeerKey
 	d.mu.RUnlock()
@@ -278,19 +277,6 @@ func (d *DnsInterceptor) writeDNSError(w dns.ResponseWriter, r *dns.Msg, logger
 	}
 }
 
-// continueToNextHandler signals the handler chain to try the next handler
-func (d *DnsInterceptor) continueToNextHandler(w dns.ResponseWriter, r *dns.Msg, logger *log.Entry, reason string) {
-	logger.Tracef("continuing to next handler for domain=%s reason=%s", r.Question[0].Name, reason)
-
-	resp := new(dns.Msg)
-	resp.SetRcode(r, dns.RcodeNameError)
-	// Set Zero bit to signal handler chain to continue
-	resp.MsgHdr.Zero = true
-	if err := w.WriteMsg(resp); err != nil {
-		logger.Errorf("failed writing DNS continue response: %v", err)
-	}
-}
-
 func (d *DnsInterceptor) getUpstreamIP(peerKey string) (netip.Addr, error) {
 	peerAllowedIP, exists := d.peerStore.AllowedIP(peerKey)
 	if !exists {

client/ios/NetBirdSDK/client.go

@@ -320,6 +320,16 @@ func (c *Client) RemoveConnectionListener() {
 	c.recorder.RemoveConnectionListener()
 }
 
+// IsLoginRequiredCached reports whether the LAST observed management error was an
+// auth failure (PermissionDenied/InvalidArgument), using the in-memory status
+// recorder. Unlike IsLoginRequired() it performs NO network call, so it is safe to
+// call from the connection listener during teardown (e.g. onDisconnected) without
+// blocking on a slow or unavailable network. Returns false while connected to
+// management or when the last error was not auth-related.
+func (c *Client) IsLoginRequiredCached() bool {
+	return c.recorder.IsLoginRequired()
+}
+
 func (c *Client) IsLoginRequired() bool {
 	var ctx context.Context
 	//nolint

flow/proto/flow.pb.go

@@ -134,11 +134,9 @@ type FlowEvent struct {
 	// When the event occurred
 	Timestamp *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
 	// Public key of the sending peer
-	PublicKey   []byte                 `protobuf:"bytes,3,opt,name=public_key,json=publicKey,proto3" json:"public_key,omitempty"`
-	FlowFields  *FlowFields            `protobuf:"bytes,4,opt,name=flow_fields,json=flowFields,proto3" json:"flow_fields,omitempty"`
-	IsInitiator bool                   `protobuf:"varint,5,opt,name=isInitiator,proto3" json:"isInitiator,omitempty"`
-	WindowStart *timestamppb.Timestamp `protobuf:"bytes,6,opt,name=window_start,json=windowStart,proto3" json:"window_start,omitempty"`
-	WindowEnd   *timestamppb.Timestamp `protobuf:"bytes,7,opt,name=window_end,json=windowEnd,proto3" json:"window_end,omitempty"`
+	PublicKey   []byte      `protobuf:"bytes,3,opt,name=public_key,json=publicKey,proto3" json:"public_key,omitempty"`
+	FlowFields  *FlowFields `protobuf:"bytes,4,opt,name=flow_fields,json=flowFields,proto3" json:"flow_fields,omitempty"`
+	IsInitiator bool        `protobuf:"varint,5,opt,name=isInitiator,proto3" json:"isInitiator,omitempty"`
 }
 
 func (x *FlowEvent) Reset() {
@@ -208,20 +206,6 @@ func (x *FlowEvent) GetIsInitiator() bool {
 	return false
 }
 
-func (x *FlowEvent) GetWindowStart() *timestamppb.Timestamp {
-	if x != nil {
-		return x.WindowStart
-	}
-	return nil
-}
-
-func (x *FlowEvent) GetWindowEnd() *timestamppb.Timestamp {
-	if x != nil {
-		return x.WindowEnd
-	}
-	return nil
-}
-
 type FlowEventAck struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -300,6 +284,7 @@ type FlowFields struct {
 	// Layer 4 -specific information
 	//
 	// Types that are assignable to ConnectionInfo:
+	//
 	//	*FlowFields_PortInfo
 	//	*FlowFields_IcmpInfo
 	ConnectionInfo isFlowFields_ConnectionInfo `protobuf_oneof:"connection_info"`
@@ -312,9 +297,6 @@ type FlowFields struct {
 	// Resource ID
 	SourceResourceId []byte `protobuf:"bytes,14,opt,name=source_resource_id,json=sourceResourceId,proto3" json:"source_resource_id,omitempty"`
 	DestResourceId   []byte `protobuf:"bytes,15,opt,name=dest_resource_id,json=destResourceId,proto3" json:"dest_resource_id,omitempty"`
-	NumOfStarts      uint64 `protobuf:"varint,16,opt,name=num_of_starts,json=numOfStarts,proto3" json:"num_of_starts,omitempty"`
-	NumOfEnds        uint64 `protobuf:"varint,17,opt,name=num_of_ends,json=numOfEnds,proto3" json:"num_of_ends,omitempty"`
-	NumOfDrops       uint64 `protobuf:"varint,18,opt,name=num_of_drops,json=numOfDrops,proto3" json:"num_of_drops,omitempty"`
 }
 
 func (x *FlowFields) Reset() {
@@ -461,27 +443,6 @@ func (x *FlowFields) GetDestResourceId() []byte {
 	return nil
 }
 
-func (x *FlowFields) GetNumOfStarts() uint64 {
-	if x != nil {
-		return x.NumOfStarts
-	}
-	return 0
-}
-
-func (x *FlowFields) GetNumOfEnds() uint64 {
-	if x != nil {
-		return x.NumOfEnds
-	}
-	return 0
-}
-
-func (x *FlowFields) GetNumOfDrops() uint64 {
-	if x != nil {
-		return x.NumOfDrops
-	}
-	return 0
-}
-
 type isFlowFields_ConnectionInfo interface {
 	isFlowFields_ConnectionInfo()
 }
@@ -618,7 +579,7 @@ var file_flow_proto_rawDesc = []byte{
 	0x0a, 0x0a, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x04, 0x66, 0x6c,
 	0x6f, 0x77, 0x1a, 0x1f, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f,
 	0x62, 0x75, 0x66, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x22, 0xce, 0x02, 0x0a, 0x09, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e,
+	0x6f, 0x74, 0x6f, 0x22, 0xd4, 0x01, 0x0a, 0x09, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e,
 	0x74, 0x12, 0x19, 0x0a, 0x08, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20,
 	0x01, 0x28, 0x0c, 0x52, 0x07, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x49, 0x64, 0x12, 0x38, 0x0a, 0x09,
 	0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32,
@@ -631,59 +592,45 @@ var file_flow_proto_rawDesc = []byte{
 	0x77, 0x2e, 0x46, 0x6c, 0x6f, 0x77, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x52, 0x0a, 0x66, 0x6c,
 	0x6f, 0x77, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x12, 0x20, 0x0a, 0x0b, 0x69, 0x73, 0x49, 0x6e,
 	0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x18, 0x05, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0b, 0x69,
-	0x73, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x12, 0x3d, 0x0a, 0x0c, 0x77, 0x69,
-	0x6e, 0x64, 0x6f, 0x77, 0x5f, 0x73, 0x74, 0x61, 0x72, 0x74, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0b,
-	0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62,
-	0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0b, 0x77, 0x69,
-	0x6e, 0x64, 0x6f, 0x77, 0x53, 0x74, 0x61, 0x72, 0x74, 0x12, 0x39, 0x0a, 0x0a, 0x77, 0x69, 0x6e,
-	0x64, 0x6f, 0x77, 0x5f, 0x65, 0x6e, 0x64, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e,
-	0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e,
-	0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x09, 0x77, 0x69, 0x6e, 0x64, 0x6f,
-	0x77, 0x45, 0x6e, 0x64, 0x22, 0x4b, 0x0a, 0x0c, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e,
-	0x74, 0x41, 0x63, 0x6b, 0x12, 0x19, 0x0a, 0x08, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x5f, 0x69, 0x64,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x49, 0x64, 0x12,
-	0x20, 0x0a, 0x0b, 0x69, 0x73, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x18, 0x02,
-	0x20, 0x01, 0x28, 0x08, 0x52, 0x0b, 0x69, 0x73, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x74, 0x6f,
-	0x72, 0x22, 0x82, 0x05, 0x0a, 0x0a, 0x46, 0x6c, 0x6f, 0x77, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x73,
-	0x12, 0x17, 0x0a, 0x07, 0x66, 0x6c, 0x6f, 0x77, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28,
-	0x0c, 0x52, 0x06, 0x66, 0x6c, 0x6f, 0x77, 0x49, 0x64, 0x12, 0x1e, 0x0a, 0x04, 0x74, 0x79, 0x70,
-	0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0a, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x54,
-	0x79, 0x70, 0x65, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x17, 0x0a, 0x07, 0x72, 0x75, 0x6c,
-	0x65, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x72, 0x75, 0x6c, 0x65,
-	0x49, 0x64, 0x12, 0x2d, 0x0a, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18,
-	0x04, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0f, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x44, 0x69, 0x72,
-	0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f,
-	0x6e, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x18, 0x05, 0x20,
-	0x01, 0x28, 0x0d, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x12, 0x1b, 0x0a,
-	0x09, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x70, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0c,
-	0x52, 0x08, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x70, 0x12, 0x17, 0x0a, 0x07, 0x64, 0x65,
-	0x73, 0x74, 0x5f, 0x69, 0x70, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x64, 0x65, 0x73,
-	0x74, 0x49, 0x70, 0x12, 0x2d, 0x0a, 0x09, 0x70, 0x6f, 0x72, 0x74, 0x5f, 0x69, 0x6e, 0x66, 0x6f,
-	0x18, 0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x50, 0x6f,
-	0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08, 0x70, 0x6f, 0x72, 0x74, 0x49, 0x6e,
-	0x66, 0x6f, 0x12, 0x2d, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x18,
-	0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x49, 0x43, 0x4d,
-	0x50, 0x49, 0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08, 0x69, 0x63, 0x6d, 0x70, 0x49, 0x6e, 0x66,
-	0x6f, 0x12, 0x1d, 0x0a, 0x0a, 0x72, 0x78, 0x5f, 0x70, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x18,
-	0x0a, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x72, 0x78, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x73,
-	0x12, 0x1d, 0x0a, 0x0a, 0x74, 0x78, 0x5f, 0x70, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x18, 0x0b,
-	0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x74, 0x78, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x12,
-	0x19, 0x0a, 0x08, 0x72, 0x78, 0x5f, 0x62, 0x79, 0x74, 0x65, 0x73, 0x18, 0x0c, 0x20, 0x01, 0x28,
-	0x04, 0x52, 0x07, 0x72, 0x78, 0x42, 0x79, 0x74, 0x65, 0x73, 0x12, 0x19, 0x0a, 0x08, 0x74, 0x78,
-	0x5f, 0x62, 0x79, 0x74, 0x65, 0x73, 0x18, 0x0d, 0x20, 0x01, 0x28, 0x04, 0x52, 0x07, 0x74, 0x78,
-	0x42, 0x79, 0x74, 0x65, 0x73, 0x12, 0x2c, 0x0a, 0x12, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f,
-	0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x0e, 0x20, 0x01, 0x28,
-	0x0c, 0x52, 0x10, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63,
-	0x65, 0x49, 0x64, 0x12, 0x28, 0x0a, 0x10, 0x64, 0x65, 0x73, 0x74, 0x5f, 0x72, 0x65, 0x73, 0x6f,
-	0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x0f, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0e, 0x64,
-	0x65, 0x73, 0x74, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x64, 0x12, 0x22, 0x0a,
-	0x0d, 0x6e, 0x75, 0x6d, 0x5f, 0x6f, 0x66, 0x5f, 0x73, 0x74, 0x61, 0x72, 0x74, 0x73, 0x18, 0x10,
-	0x20, 0x01, 0x28, 0x04, 0x52, 0x0b, 0x6e, 0x75, 0x6d, 0x4f, 0x66, 0x53, 0x74, 0x61, 0x72, 0x74,
-	0x73, 0x12, 0x1e, 0x0a, 0x0b, 0x6e, 0x75, 0x6d, 0x5f, 0x6f, 0x66, 0x5f, 0x65, 0x6e, 0x64, 0x73,
-	0x18, 0x11, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x6e, 0x75, 0x6d, 0x4f, 0x66, 0x45, 0x6e, 0x64,
-	0x73, 0x12, 0x20, 0x0a, 0x0c, 0x6e, 0x75, 0x6d, 0x5f, 0x6f, 0x66, 0x5f, 0x64, 0x72, 0x6f, 0x70,
-	0x73, 0x18, 0x12, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0a, 0x6e, 0x75, 0x6d, 0x4f, 0x66, 0x44, 0x72,
-	0x6f, 0x70, 0x73, 0x42, 0x11, 0x0a, 0x0f, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f,
+	0x73, 0x49, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x22, 0x4b, 0x0a, 0x0c, 0x46, 0x6c,
+	0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x41, 0x63, 0x6b, 0x12, 0x19, 0x0a, 0x08, 0x65, 0x76,
+	0x65, 0x6e, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x65, 0x76,
+	0x65, 0x6e, 0x74, 0x49, 0x64, 0x12, 0x20, 0x0a, 0x0b, 0x69, 0x73, 0x49, 0x6e, 0x69, 0x74, 0x69,
+	0x61, 0x74, 0x6f, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0b, 0x69, 0x73, 0x49, 0x6e,
+	0x69, 0x74, 0x69, 0x61, 0x74, 0x6f, 0x72, 0x22, 0x9c, 0x04, 0x0a, 0x0a, 0x46, 0x6c, 0x6f, 0x77,
+	0x46, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x12, 0x17, 0x0a, 0x07, 0x66, 0x6c, 0x6f, 0x77, 0x5f, 0x69,
+	0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x66, 0x6c, 0x6f, 0x77, 0x49, 0x64, 0x12,
+	0x1e, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0a, 0x2e,
+	0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12,
+	0x17, 0x0a, 0x07, 0x72, 0x75, 0x6c, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c,
+	0x52, 0x06, 0x72, 0x75, 0x6c, 0x65, 0x49, 0x64, 0x12, 0x2d, 0x0a, 0x09, 0x64, 0x69, 0x72, 0x65,
+	0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0f, 0x2e, 0x66, 0x6c,
+	0x6f, 0x77, 0x2e, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x64, 0x69,
+	0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x63, 0x6f, 0x6c, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x63, 0x6f, 0x6c, 0x12, 0x1b, 0x0a, 0x09, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x70,
+	0x18, 0x06, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x08, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x70,
+	0x12, 0x17, 0x0a, 0x07, 0x64, 0x65, 0x73, 0x74, 0x5f, 0x69, 0x70, 0x18, 0x07, 0x20, 0x01, 0x28,
+	0x0c, 0x52, 0x06, 0x64, 0x65, 0x73, 0x74, 0x49, 0x70, 0x12, 0x2d, 0x0a, 0x09, 0x70, 0x6f, 0x72,
+	0x74, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x18, 0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66,
+	0x6c, 0x6f, 0x77, 0x2e, 0x50, 0x6f, 0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08,
+	0x70, 0x6f, 0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x2d, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70,
+	0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66, 0x6c,
+	0x6f, 0x77, 0x2e, 0x49, 0x43, 0x4d, 0x50, 0x49, 0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08, 0x69,
+	0x63, 0x6d, 0x70, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x1d, 0x0a, 0x0a, 0x72, 0x78, 0x5f, 0x70, 0x61,
+	0x63, 0x6b, 0x65, 0x74, 0x73, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x72, 0x78, 0x50,
+	0x61, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x12, 0x1d, 0x0a, 0x0a, 0x74, 0x78, 0x5f, 0x70, 0x61, 0x63,
+	0x6b, 0x65, 0x74, 0x73, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x74, 0x78, 0x50, 0x61,
+	0x63, 0x6b, 0x65, 0x74, 0x73, 0x12, 0x19, 0x0a, 0x08, 0x72, 0x78, 0x5f, 0x62, 0x79, 0x74, 0x65,
+	0x73, 0x18, 0x0c, 0x20, 0x01, 0x28, 0x04, 0x52, 0x07, 0x72, 0x78, 0x42, 0x79, 0x74, 0x65, 0x73,
+	0x12, 0x19, 0x0a, 0x08, 0x74, 0x78, 0x5f, 0x62, 0x79, 0x74, 0x65, 0x73, 0x18, 0x0d, 0x20, 0x01,
+	0x28, 0x04, 0x52, 0x07, 0x74, 0x78, 0x42, 0x79, 0x74, 0x65, 0x73, 0x12, 0x2c, 0x0a, 0x12, 0x73,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69,
+	0x64, 0x18, 0x0e, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x10, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52,
+	0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x64, 0x12, 0x28, 0x0a, 0x10, 0x64, 0x65, 0x73,
+	0x74, 0x5f, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x0f, 0x20,
+	0x01, 0x28, 0x0c, 0x52, 0x0e, 0x64, 0x65, 0x73, 0x74, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63,
+	0x65, 0x49, 0x64, 0x42, 0x11, 0x0a, 0x0f, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f,
 	0x6e, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x22, 0x48, 0x0a, 0x08, 0x50, 0x6f, 0x72, 0x74, 0x49, 0x6e,
 	0x66, 0x6f, 0x12, 0x1f, 0x0a, 0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x70, 0x6f, 0x72,
 	0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0a, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x50,
@@ -736,19 +683,17 @@ var file_flow_proto_goTypes = []interface{}{
 var file_flow_proto_depIdxs = []int32{
 	7, // 0: flow.FlowEvent.timestamp:type_name -> google.protobuf.Timestamp
 	4, // 1: flow.FlowEvent.flow_fields:type_name -> flow.FlowFields
-	7, // 2: flow.FlowEvent.window_start:type_name -> google.protobuf.Timestamp
-	7, // 3: flow.FlowEvent.window_end:type_name -> google.protobuf.Timestamp
-	0, // 4: flow.FlowFields.type:type_name -> flow.Type
-	1, // 5: flow.FlowFields.direction:type_name -> flow.Direction
-	5, // 6: flow.FlowFields.port_info:type_name -> flow.PortInfo
-	6, // 7: flow.FlowFields.icmp_info:type_name -> flow.ICMPInfo
-	2, // 8: flow.FlowService.Events:input_type -> flow.FlowEvent
-	3, // 9: flow.FlowService.Events:output_type -> flow.FlowEventAck
-	9, // [9:10] is the sub-list for method output_type
-	8, // [8:9] is the sub-list for method input_type
-	8, // [8:8] is the sub-list for extension type_name
-	8, // [8:8] is the sub-list for extension extendee
-	0, // [0:8] is the sub-list for field type_name
+	0, // 2: flow.FlowFields.type:type_name -> flow.Type
+	1, // 3: flow.FlowFields.direction:type_name -> flow.Direction
+	5, // 4: flow.FlowFields.port_info:type_name -> flow.PortInfo
+	6, // 5: flow.FlowFields.icmp_info:type_name -> flow.ICMPInfo
+	2, // 6: flow.FlowService.Events:input_type -> flow.FlowEvent
+	3, // 7: flow.FlowService.Events:output_type -> flow.FlowEventAck
+	7, // [7:8] is the sub-list for method output_type
+	6, // [6:7] is the sub-list for method input_type
+	6, // [6:6] is the sub-list for extension type_name
+	6, // [6:6] is the sub-list for extension extendee
+	0, // [0:6] is the sub-list for field type_name
 }
 
 func init() { file_flow_proto_init() }

flow/proto/flow.proto

@@ -24,9 +24,6 @@ message FlowEvent {
   FlowFields flow_fields = 4;
 
   bool isInitiator = 5;
-
-  google.protobuf.Timestamp window_start = 6;
-  google.protobuf.Timestamp window_end = 7;
 }
 
 message FlowEventAck {
@@ -78,9 +75,6 @@ message FlowFields {
   bytes source_resource_id = 14;
   bytes dest_resource_id = 15;
 
-  uint64 num_of_starts = 16;
-  uint64 num_of_ends = 17;
-  uint64 num_of_drops = 18;
 }
 
 // Flow event types

flow/proto/flow_grpc.pb.go

@@ -1,8 +1,4 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
-// versions:
-// - protoc-gen-go-grpc v1.6.1
-// - protoc             v3.21.9
-// source: flow.proto
 
 package proto
 
@@ -15,19 +11,15 @@ import (
 
 // This is a compile-time assertion to ensure that this generated file
 // is compatible with the grpc package it is being compiled against.
-// Requires gRPC-Go v1.64.0 or later.
-const _ = grpc.SupportPackageIsVersion9
-
-const (
-	FlowService_Events_FullMethodName = "/flow.FlowService/Events"
-)
+// Requires gRPC-Go v1.32.0 or later.
+const _ = grpc.SupportPackageIsVersion7
 
 // FlowServiceClient is the client API for FlowService service.
 //
 // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
 type FlowServiceClient interface {
 	// Client to receiver streams of events and acknowledgements
-	Events(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[FlowEvent, FlowEventAck], error)
+	Events(ctx context.Context, opts ...grpc.CallOption) (FlowService_EventsClient, error)
 }
 
 type flowServiceClient struct {
@@ -38,40 +30,54 @@ func NewFlowServiceClient(cc grpc.ClientConnInterface) FlowServiceClient {
 	return &flowServiceClient{cc}
 }
 
-func (c *flowServiceClient) Events(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[FlowEvent, FlowEventAck], error) {
-	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	stream, err := c.cc.NewStream(ctx, &FlowService_ServiceDesc.Streams[0], FlowService_Events_FullMethodName, cOpts...)
+func (c *flowServiceClient) Events(ctx context.Context, opts ...grpc.CallOption) (FlowService_EventsClient, error) {
+	stream, err := c.cc.NewStream(ctx, &FlowService_ServiceDesc.Streams[0], "/flow.FlowService/Events", opts...)
 	if err != nil {
 		return nil, err
 	}
-	x := &grpc.GenericClientStream[FlowEvent, FlowEventAck]{ClientStream: stream}
+	x := &flowServiceEventsClient{stream}
 	return x, nil
 }
 
-// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
-type FlowService_EventsClient = grpc.BidiStreamingClient[FlowEvent, FlowEventAck]
+type FlowService_EventsClient interface {
+	Send(*FlowEvent) error
+	Recv() (*FlowEventAck, error)
+	grpc.ClientStream
+}
+
+type flowServiceEventsClient struct {
+	grpc.ClientStream
+}
+
+func (x *flowServiceEventsClient) Send(m *FlowEvent) error {
+	return x.ClientStream.SendMsg(m)
+}
+
+func (x *flowServiceEventsClient) Recv() (*FlowEventAck, error) {
+	m := new(FlowEventAck)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
 
 // FlowServiceServer is the server API for FlowService service.
 // All implementations must embed UnimplementedFlowServiceServer
-// for forward compatibility.
+// for forward compatibility
 type FlowServiceServer interface {
 	// Client to receiver streams of events and acknowledgements
-	Events(grpc.BidiStreamingServer[FlowEvent, FlowEventAck]) error
+	Events(FlowService_EventsServer) error
 	mustEmbedUnimplementedFlowServiceServer()
 }
 
-// UnimplementedFlowServiceServer must be embedded to have
-// forward compatible implementations.
-//
-// NOTE: this should be embedded by value instead of pointer to avoid a nil
-// pointer dereference when methods are called.
-type UnimplementedFlowServiceServer struct{}
+// UnimplementedFlowServiceServer must be embedded to have forward compatible implementations.
+type UnimplementedFlowServiceServer struct {
+}
 
-func (UnimplementedFlowServiceServer) Events(grpc.BidiStreamingServer[FlowEvent, FlowEventAck]) error {
-	return status.Error(codes.Unimplemented, "method Events not implemented")
+func (UnimplementedFlowServiceServer) Events(FlowService_EventsServer) error {
+	return status.Errorf(codes.Unimplemented, "method Events not implemented")
 }
 func (UnimplementedFlowServiceServer) mustEmbedUnimplementedFlowServiceServer() {}
-func (UnimplementedFlowServiceServer) testEmbeddedByValue()                     {}
 
 // UnsafeFlowServiceServer may be embedded to opt out of forward compatibility for this service.
 // Use of this interface is not recommended, as added methods to FlowServiceServer will
@@ -81,22 +87,34 @@ type UnsafeFlowServiceServer interface {
 }
 
 func RegisterFlowServiceServer(s grpc.ServiceRegistrar, srv FlowServiceServer) {
-	// If the following call panics, it indicates UnimplementedFlowServiceServer was
-	// embedded by pointer and is nil.  This will cause panics if an
-	// unimplemented method is ever invoked, so we test this at initialization
-	// time to prevent it from happening at runtime later due to I/O.
-	if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
-		t.testEmbeddedByValue()
-	}
 	s.RegisterService(&FlowService_ServiceDesc, srv)
 }
 
 func _FlowService_Events_Handler(srv interface{}, stream grpc.ServerStream) error {
-	return srv.(FlowServiceServer).Events(&grpc.GenericServerStream[FlowEvent, FlowEventAck]{ServerStream: stream})
+	return srv.(FlowServiceServer).Events(&flowServiceEventsServer{stream})
 }
 
-// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
-type FlowService_EventsServer = grpc.BidiStreamingServer[FlowEvent, FlowEventAck]
+type FlowService_EventsServer interface {
+	Send(*FlowEventAck) error
+	Recv() (*FlowEvent, error)
+	grpc.ServerStream
+}
+
+type flowServiceEventsServer struct {
+	grpc.ServerStream
+}
+
+func (x *flowServiceEventsServer) Send(m *FlowEventAck) error {
+	return x.ServerStream.SendMsg(m)
+}
+
+func (x *flowServiceEventsServer) Recv() (*FlowEvent, error) {
+	m := new(FlowEvent)
+	if err := x.ServerStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
 
 // FlowService_ServiceDesc is the grpc.ServiceDesc for FlowService service.
 // It's only intended for direct use with grpc.RegisterService,

flow/proto/generate.sh

@@ -10,7 +10,6 @@ fi
 
 old_pwd=$(pwd)
 script_path=$(dirname $(realpath "$0"))
-echo "$script_path"
 cd "$script_path"
 go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.26
 go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.1

shared/management/http/api/openapi.yml

@@ -2765,28 +2765,6 @@ components:
           type: integer
           description: "Number of packets transmitted."
           example: 5
-        num_of_starts:
-          type: integer
-          description: "Number of start events."
-          example: 3
-        num_of_ends:
-          type: integer
-          description: "Number of end events."
-          example: 4
-        num_of_drops:
-          type: integer
-          description: "Number of drop events."
-          example: 5
-        window_start:
-          type: string
-          format: date-time
-          description: Timestamp of the start of the aggregation window.
-          example: 2025-03-20T16:23:58.125397Z
-        window_end:
-          type: string
-          format: date-time
-          description: Timestamp of the end of the aggregation window.
-          example: 2025-03-20T16:23:58.125397Z
         events:
           type: array
           description: "List of events that are correlated to this flow (e.g., start, end)."
@@ -2808,11 +2786,6 @@ components:
         - rx_packets
         - tx_bytes
         - tx_packets
-        - num_of_starts
-        - num_of_ends
-        - num_of_drops
-        - window_start
-        - window_end
         - events
     NetworkTrafficEventsResponse:
       type: object

shared/management/http/api/types.gen.go

@@ -2905,18 +2905,9 @@ type NetworkTrafficEvent struct {
 	Events []NetworkTrafficSubEvent `json:"events"`
 
 	// FlowId FlowID is the ID of the connection flow. Not unique because it can be the same for multiple events (e.g., start and end of the connection).
-	FlowId string             `json:"flow_id"`
-	Icmp   NetworkTrafficICMP `json:"icmp"`
-
-	// NumOfDrops Number of drop events.
-	NumOfDrops int `json:"num_of_drops"`
-
-	// NumOfEnds Number of end events.
-	NumOfEnds int `json:"num_of_ends"`
-
-	// NumOfStarts Number of start events.
-	NumOfStarts int                  `json:"num_of_starts"`
-	Policy      NetworkTrafficPolicy `json:"policy"`
+	FlowId string               `json:"flow_id"`
+	Icmp   NetworkTrafficICMP   `json:"icmp"`
+	Policy NetworkTrafficPolicy `json:"policy"`
 
 	// Protocol Protocol is the protocol of the traffic (e.g. 1 = ICMP, 6 = TCP, 17 = UDP, etc.).
 	Protocol int `json:"protocol"`
@@ -2937,12 +2928,6 @@ type NetworkTrafficEvent struct {
 	// TxPackets Number of packets transmitted.
 	TxPackets int                `json:"tx_packets"`
 	User      NetworkTrafficUser `json:"user"`
-
-	// WindowEnd Timestamp of the end of the aggregation window.
-	WindowEnd time.Time `json:"window_end"`
-
-	// WindowStart Timestamp of the start of the aggregation window.
-	WindowStart time.Time `json:"window_start"`
 }
 
 // NetworkTrafficEventsResponse defines model for NetworkTrafficEventsResponse.