new kill session by authid

Author: Marko Petzold

router/killbyauthid_test.go

@@ -0,0 +1,49 @@
+package router //nolint:testpackage
+
+import (
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/gammazero/nexus/v3/wamp"
+)
+
+// TestKillSessionsByAuthidGoAPI covers the Go-level Router.KillSessionsByAuthid
+// added for embedders (the cluster mesh) to enforce single-authority
+// identities: it closes matching sessions except the excluded one, returns the
+// kill count, and reports -1 for an unknown realm.
+func TestKillSessionsByAuthidGoAPI(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		r := newTestRouter(t)
+		defer r.Close()
+
+		// testClient gives every session authid "user1".
+		keep := testClient(t, r)
+		victim1 := testClient(t, r)
+		victim2 := testClient(t, r)
+
+		// Unknown realm → -1, nothing killed.
+		require.Equal(t, -1, r.KillSessionsByAuthid("no.such.realm", "user1", 0))
+
+		// Kill all "user1" sessions except `keep`.
+		killed := r.KillSessionsByAuthid(testRealm, "user1", keep.ID)
+		require.Equal(t, 2, killed, "expected both non-excluded sessions killed")
+
+		for i, v := range []*wamp.Session{victim1, victim2} {
+			msg, err := wamp.RecvTimeout(v, time.Second)
+			require.NoErrorf(t, err, "victim %d should receive GOODBYE", i+1)
+			g, ok := msg.(*wamp.Goodbye)
+			require.Truef(t, ok, "victim %d expected GOODBYE", i+1)
+			require.Equal(t, wamp.CloseNormal, g.Reason)
+		}
+
+		// The excluded session is untouched.
+		_, err := wamp.RecvTimeout(keep, time.Millisecond)
+		require.Error(t, err, "excluded session must not be killed")
+
+		// A non-matching authid kills nothing.
+		require.Equal(t, 0, r.KillSessionsByAuthid(testRealm, "nobody", 0))
+	})
+}

router/router.go

@@ -54,6 +54,15 @@ type Router interface {
 	// is used as a pluggable library sometimes it is usefully to expose WAMP
 	// features current version provides.
 	RouterFeatures() *wamp.Dict
+
+	// KillSessionsByAuthid closes every session on the named realm whose
+	// authid equals the given value, except the session identified by
+	// exclude (pass 0 to exclude none). It returns the number of sessions
+	// closed, or -1 if the realm is not served by this router. This is the
+	// Go-level equivalent of the wamp.session.kill_by_authid meta procedure,
+	// exposed so an embedder (e.g. the cluster mesh) can enforce
+	// single-authority identities without standing up a meta caller.
+	KillSessionsByAuthid(realm wamp.URI, authid string, exclude wamp.ID) int
 }
 
 // router is the default WAMP router implementation.
@@ -74,6 +83,26 @@ type router struct {
 	memStatsStopped chan struct{}
 }
 
+// KillSessionsByAuthid closes every session on realmURI whose authid matches,
+// except exclude (0 = none). Returns the kill count, or -1 if the realm is not
+// served here. The realm is located under the router actor; the kill itself
+// runs on the realm actor (killSessionsByDetail self-dispatches), so this never
+// holds both actors at once.
+func (r *router) KillSessionsByAuthid(realmURI wamp.URI, authid string, exclude wamp.ID) int {
+	var realm *realm
+	sync := make(chan struct{})
+	r.actionChan <- func() {
+		realm = r.realms[realmURI]
+		close(sync)
+	}
+	<-sync
+	if realm == nil {
+		return -1
+	}
+	return realm.killSessionsByDetail("authid", authid, wamp.CloseNormal,
+		"superseded by a newer session with the same authid", exclude)
+}
+
 // NewRouter creates a WAMP router instance.
 func NewRouter(config *Config, logger stdlog.StdLog) (Router, error) {
 	// If logger not provided, create one.