internal/uevent: consume kernel-originated events

These are not possible to cover outside integration vm. Extreme care is required when dealing with this method, so keep it simple.

Signed-off-by: Ophestra <cat@gensokyo.uk>

internal/uevent/uevent_test.go

@@ -1,9 +1,14 @@
 package uevent_test
 
 import (
+	"context"
 	"encoding"
+	"os"
 	"reflect"
+	"sync"
+	"syscall"
 	"testing"
+	"time"
 
 	"hakurei.app/internal/uevent"
 )
@@ -108,6 +113,92 @@ func adeB[V any, S interface {
 	}
 }
 
+func TestDialConsume(t *testing.T) {
+	t.Parallel()
+
+	c, err := uevent.Dial()
+	if err != nil {
+		t.Fatalf("Dial: error = %v", err)
+	}
+	t.Cleanup(func() {
+		if closeErr := c.Close(); closeErr != nil {
+			t.Fatal(err)
+		}
+	})
+
+	// check kernel-assigned port id
+	c0, err0 := uevent.Dial()
+	if err0 != nil {
+		t.Fatalf("Dial: error = %v", err)
+	}
+	t.Cleanup(func() {
+		if closeErr := c0.Close(); closeErr != nil {
+			t.Fatal(closeErr)
+		}
+	})
+
+	var wg sync.WaitGroup
+	done := make(chan struct{})
+	events := make(chan *uevent.Message, 1<<10)
+	go func() {
+		defer close(done)
+		for msg := range events {
+			t.Log(msg)
+		}
+	}()
+	t.Cleanup(func() {
+		wg.Wait()
+		close(events)
+		<-done
+	})
+
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+
+	wg.Go(func() {
+		if err = c.Consume(ctx, events); err != context.Canceled {
+			panic(err)
+		}
+	})
+	wg.Go(func() {
+		if err0 = c0.Consume(ctx, events); err0 != context.Canceled {
+			panic(err0)
+		}
+	})
+
+	if testing.Verbose() {
+		if d, perr := time.ParseDuration(os.Getenv(
+			"ROSA_UEVENT_TEST_DURATION",
+		)); perr != nil {
+			t.Logf("skipping long test: error = %v", perr)
+		} else {
+			time.Sleep(d)
+		}
+	}
+	cancel()
+	wg.Wait()
+
+	ctx, cancel = context.WithCancel(t.Context())
+	defer cancel()
+
+	var errs [2]error
+	exclExit := make(chan struct{})
+	wg.Go(func() {
+		defer func() { exclExit <- struct{}{} }()
+		errs[0] = c.Consume(ctx, events)
+	})
+	wg.Go(func() {
+		defer func() { exclExit <- struct{}{} }()
+		errs[1] = c.Consume(ctx, events)
+	})
+	<-exclExit
+	cancel()
+	<-exclExit
+	if errs[0] != syscall.EAGAIN && errs[1] != syscall.EAGAIN {
+		t.Fatalf("enterExcl: err0 = %v, err1 = %v", errs[0], errs[1])
+	}
+}
+
 func TestErrors(t *testing.T) {
 	t.Parallel()
 
@@ -138,6 +229,10 @@ func TestErrors(t *testing.T) {
 			Data: "\x00",
 			Kind: 0xbad,
 		}, `section "" is invalid`},
+
+		{"BadPortError", &uevent.BadPortError{
+			Pid: 1,
+		}, "unexpected message from port id 1 on NETLINK_KOBJECT_UEVENT"},
 	}
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {