#75 program BPF maps before attach to eliminate startup-race drops

cmd/github_actions_handler.go

@@ -77,7 +77,10 @@ func (h *GitHubActionsHandler) Handle(_ context.Context, r slog.Record) error {
 		return true
 	})
 
-	fmt.Fprintf(os.Stderr, "%s%s\n", prefix, sb.String())
+	// Leading timestamp on every line. Placed after the prefix so GitHub still
+	// parses the ::error::/::warning:: workflow command at column 0. Millisecond
+	// precision because the startup race this surfaces is sub-second.
+	fmt.Fprintf(os.Stderr, "%s%s %s\n", prefix, r.Time.Format("2006-01-02 15:04:05.000"), sb.String())
 	return nil
 }
 

cmd/github_actions_handler_test.go

@@ -0,0 +1,75 @@
+//   Copyright 2026 BoxBuild Inc DBA CodeCargo
+//
+//   Licensed under the Apache License, Version 2.0 (the "License");
+//   you may not use this file except in compliance with the License.
+//   You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+//   Unless required by applicable law or agreed to in writing, software
+//   distributed under the License is distributed on an "AS IS" BASIS,
+//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//   See the License for the specific language governing permissions and
+//   limitations under the License.
+
+//go:build linux
+
+package cmd
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// captureStderr redirects os.Stderr around fn and returns what was written. The
+// GitHubActionsHandler writes to os.Stderr directly, so the test swaps the global
+// for the duration of the call.
+func captureStderr(t *testing.T, fn func()) string {
+	t.Helper()
+	old := os.Stderr
+	r, w, err := os.Pipe()
+	require.NoError(t, err)
+	os.Stderr = w
+	defer func() { os.Stderr = old }()
+
+	fn()
+	require.NoError(t, w.Close())
+	data, err := io.ReadAll(r)
+	require.NoError(t, err)
+	return string(data)
+}
+
+// TestGitHubActionsHandler_TimestampPrefix verifies every emitted line carries a
+// leading millisecond timestamp, and that for annotated levels the ::error:: /
+// ::warning:: workflow command stays at column 0 (timestamp after the prefix).
+func TestGitHubActionsHandler_TimestampPrefix(t *testing.T) {
+	h := NewGitHubActionsHandler(false)
+	ts := time.Date(2026, 6, 29, 15, 24, 8, 456_000_000, time.UTC)
+
+	out := captureStderr(t, func() {
+		info := slog.NewRecord(ts, slog.LevelInfo, "Connection blocked", 0)
+		info.AddAttrs(slog.Bool("pre_ready", true))
+		require.NoError(t, h.Handle(context.Background(), info))
+
+		errRec := slog.NewRecord(ts, slog.LevelError, "boom", 0)
+		require.NoError(t, h.Handle(context.Background(), errRec))
+	})
+
+	lines := strings.Split(strings.TrimRight(out, "\n"), "\n")
+	require.Len(t, lines, 2)
+
+	// Info line: "<timestamp> <message> <attrs>", no prefix.
+	assert.Regexp(t, `^2026-06-29 15:24:08\.456 Connection blocked\b`, lines[0])
+	assert.Contains(t, lines[0], "pre_ready=true")
+
+	// Error line: prefix stays at column 0, timestamp follows it.
+	assert.Regexp(t, `^::error::2026-06-29 15:24:08\.456 boom\b`, lines[1])
+}

cmd/start.go

@@ -309,8 +309,14 @@ func StartCargoWall(cmd *StartCmd, hooks *StartHooks) error {
 	}
 	defer rd.Close()
 
+	// Readiness boundary for startup-race suppression: blocked events stamped
+	// (CLOCK_MONOTONIC) before MarkReady are drops from the attach-before-program
+	// window and are excluded from the audit log / SaaS / summary / notifications.
+	// MarkReady runs once, just before the ready signal is published.
+	readiness := events.NewReadiness()
+
 	// Start processors before attaching programs
-	go events.ProcessBlockedEvents(rd, configMgr, notificationTracker, auditLogger, fw, logger)
+	go events.ProcessBlockedEvents(rd, configMgr, notificationTracker, auditLogger, fw, readiness, logger)
 
 	// Attach TC programs
 	egressLink, err := tc.AttachEgress(ifname, objs.TcEgress, logger)
@@ -455,6 +461,12 @@ func StartCargoWall(cmd *StartCmd, hooks *StartHooks) error {
 		}()
 	}
 
+	// Record the readiness boundary BEFORE publishing the ready signal, so the
+	// happens-before is: boundary recorded → sentinel/hook fires → CI proceeds →
+	// user traffic. A user-triggered block is therefore stamped after the
+	// boundary and correctly reported; it can never be misclassified as pre-ready.
+	readiness.MarkReady()
+
 	if hooks != nil && hooks.Ready != nil {
 		if err := hooks.Ready(); err != nil {
 			return fmt.Errorf("ready hook failed: %w", err)

pkg/events/events.go

@@ -25,6 +25,7 @@ import (
 	"os"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 	"unsafe"
 
@@ -40,6 +41,62 @@ type FirewallUpdater interface {
 	AddIP(ip net.IP, action config.Action, ports []config.Port) (bool, error)
 }
 
+// Readiness records the CLOCK_MONOTONIC instant at which CargoWall finished
+// programming its allow rules and published the ready signal. Blocked events
+// whose BPF timestamp (bpf_ktime_get_ns, also CLOCK_MONOTONIC) predates that
+// instant are startup-race drops — the TC program attaches default-deny before
+// the auto-allow infra rules land, so a connection in that window is dropped
+// once and self-heals on TCP retransmit. They are not policy decisions, so they
+// are excluded from the audit log (hence the SaaS push + summary) and from block
+// notifications. The zero value is "not ready yet": every event before MarkReady
+// is treated as pre-ready.
+type Readiness struct {
+	readyMono atomic.Uint64 // CLOCK_MONOTONIC ns at ready; 0 = not yet ready
+}
+
+// NewReadiness returns a Readiness in the not-yet-ready state.
+func NewReadiness() *Readiness { return &Readiness{} }
+
+// MarkReady records the current CLOCK_MONOTONIC instant as the readiness
+// boundary. Call exactly once, immediately before publishing the ready signal,
+// so it happens-before any externally triggered (e.g. user) traffic.
+func (r *Readiness) MarkReady() {
+	var ts unix.Timespec
+	if err := unix.ClockGettime(unix.CLOCK_MONOTONIC, &ts); err != nil {
+		// Fail open for reporting: never leave readyMono at 0 (which would
+		// suppress every block forever). A 1ns boundary suppresses nothing
+		// real — every genuine bpf_ktime_get_ns stamp is far larger.
+		r.readyMono.Store(1)
+		return
+	}
+	r.readyMono.Store(uint64(ts.Nano()))
+}
+
+// isPreReady reports whether a blocked event with the given CLOCK_MONOTONIC BPF
+// timestamp occurred before the readiness boundary (and so should be excluded
+// from reporting). Precedence:
+//   - a nil Readiness never suppresses (reporting is unaffected when the gate
+//     isn't wired up);
+//   - before MarkReady (readyMono == 0) the firewall is still being programmed,
+//     so the whole window is pre-ready and every event is suppressed — including
+//     one with a zero/garbage stamp;
+//   - after MarkReady, a zero event timestamp never suppresses, so a
+//     missing/garbage field can't mask a real (post-ready) block; otherwise the
+//     event is pre-ready iff its stamp precedes the boundary.
+func (r *Readiness) isPreReady(eventMono uint64) bool {
+	if r == nil {
+		return false
+	}
+	readyMono := r.readyMono.Load()
+	if readyMono == 0 { // MarkReady not run yet → everything so far is pre-ready
+		return true
+	}
+	if eventMono == 0 { // defensive: never suppress on a zero/garbage stamp
+		return false
+	}
+	return eventMono < readyMono
+}
+
 // lookupProcessName reads the process name from /proc/<pid>/comm.
 // Returns empty string if the process no longer exists or can't be read.
 func lookupProcessName(pid uint32) string {
@@ -244,7 +301,7 @@ func dstPortAllowedByRule(dstPort uint16, proto uint8, ports []config.Port) bool
 
 // processEvent handles a single blocked/allowed event from the ring buffer.
 func processEvent(raw []byte, configMgr *config.Manager, notificationTracker *NotificationTracker,
-	auditLogger *AuditLogger, fw FirewallUpdater, logger *slog.Logger,
+	auditLogger *AuditLogger, fw FirewallUpdater, readiness *Readiness, logger *slog.Logger,
 ) {
 	if len(raw) < int(unsafe.Sizeof(BpfBlockedEvent{})) {
 		return
@@ -452,15 +509,32 @@ func processEvent(raw []byte, configMgr *config.Manager, notificationTracker *No
 	} else if event.SrcPort == 0 && event.DstPort < 256 {
 		// Non-TCP/UDP protocol block — dst_port contains the protocol number
 		protocolName := getProtocolName(uint8(event.DstPort))
+
+		if readiness.isPreReady(event.Timestamp) {
+			// Startup-race drop (see Readiness): keep it in the process log for
+			// forensics, but exclude it from the audit log (→ SaaS push +
+			// summary) and the notification — it is not a policy decision.
+			logConnEvent(logger, "Protocol blocked", cnameChain,
+				"src", srcIP,
+				"dst", displayHostname,
+				"dst_ip", dstIP,
+				"protocol", protocolName,
+				"protocol_num", event.DstPort,
+				"process", comm,
+				"pid", pid,
+				"pre_ready", true,
+				"reported", false)
+			return
+		}
+
 		logConnEvent(logger, "Protocol blocked", cnameChain,
 			"src", srcIP,
 			"dst", displayHostname,
 			"dst_ip", dstIP,
 			"protocol", protocolName,
 			"protocol_num", event.DstPort,
 			"process", comm,
-			"pid", pid,
-			"timestamp", time.Now().Format("2006-01-02 15:04:05"))
+			"pid", pid)
 
 		// Log to audit file if configured
 		if auditLogger != nil {
@@ -496,14 +570,32 @@ func processEvent(raw []byte, configMgr *config.Manager, notificationTracker *No
 		}
 	} else {
 		// Blocked TCP SYN or UDP connection
+		if readiness.isPreReady(event.Timestamp) {
+			// Startup-race drop (see Readiness): the TC program attached
+			// default-deny before the auto-allow infra rules were programmed, so
+			// this was dropped once and self-heals on retransmit. Keep it in the
+			// process log for forensics, but exclude it from the audit log (→
+			// SaaS push + summary) and the notification — it is not a policy
+			// decision.
+			logConnEvent(logger, "Connection blocked", cnameChain,
+				"src", fmt.Sprintf("%s:%d", srcIP, event.SrcPort),
+				"dst", displayHostname,
+				"dst_ip", dstIP,
+				"dst_port", event.DstPort,
+				"process", comm,
+				"pid", pid,
+				"pre_ready", true,
+				"reported", false)
+			return
+		}
+
 		logConnEvent(logger, "Connection blocked", cnameChain,
 			"src", fmt.Sprintf("%s:%d", srcIP, event.SrcPort),
 			"dst", displayHostname,
 			"dst_ip", dstIP,
 			"dst_port", event.DstPort,
 			"process", comm,
-			"pid", pid,
-			"timestamp", time.Now().Format("2006-01-02 15:04:05"))
+			"pid", pid)
 
 		// Log to audit file if configured
 		if auditLogger != nil {
@@ -539,7 +631,7 @@ func logConnEvent(logger *slog.Logger, msg string, cnameChain []string, attrs ..
 }
 
 // ProcessBlockedEvents processes blocked connection events
-func ProcessBlockedEvents(rd *ringbuf.Reader, configMgr *config.Manager, notificationTracker *NotificationTracker, auditLogger *AuditLogger, fw FirewallUpdater, logger *slog.Logger) {
+func ProcessBlockedEvents(rd *ringbuf.Reader, configMgr *config.Manager, notificationTracker *NotificationTracker, auditLogger *AuditLogger, fw FirewallUpdater, readiness *Readiness, logger *slog.Logger) {
 	for {
 		record, err := rd.Read()
 		if err != nil {
@@ -550,6 +642,6 @@ func ProcessBlockedEvents(rd *ringbuf.Reader, configMgr *config.Manager, notific
 			continue
 		}
 
-		processEvent(record.RawSample, configMgr, notificationTracker, auditLogger, fw, logger)
+		processEvent(record.RawSample, configMgr, notificationTracker, auditLogger, fw, readiness, logger)
 	}
 }