router/dealer: implement force_reregister RegisterOption

Honor the WAMP advanced-profile force_reregister option on REGISTER:
when set to true and an existing single-policy registration is present,
forcibly evict the prior callee(s) and install the new registration
instead of returning wamp.error.procedure_already_exists.

- Advertise force_reregister=true in dealer role features.
- Add wamp.OptForceReregister option key + wamp.FeatureForceReregister.
- Add wamp.ErrUnregistered (wamp.error.unregistered) revocation reason.
- Extend Unregistered with an optional Details dict so the dealer can
  emit the unsolicited [UNREGISTERED, 0, Details] revocation form
  carrying Details.registration and Details.reason to evicted callees.
  Normal [UNREGISTERED, Request] replies stay byte-for-byte identical
  (omitempty).
- syncEvictRegistration tears down the existing registration, notifies
  every attached callee, scrubs calleeRegIDSet / procRegMap /
  registrations, and emits on_unregister (per callee) + on_delete meta
  events. The new REGISTER then falls through the fresh-registration
  branch, producing a new regID and the usual on_create + on_register
  meta events.
- force_reregister applies only when the existing registration's
  invocation policy is single (or empty); shared registrations are
  untouched, falling through to the existing policy-conflict path.

Tests:
- router/dealer_test.go: happy-path eviction, shared-reg ignore-path,
  feature-advertisement pin.
- test/spec_force_reregister_test.go: end-to-end across the transport ×
  serializer matrix; verifies subsequent CALLs route to the new callee.

Author: Marko Petzold

README.md

@@ -3,7 +3,7 @@
 # WAMP v2 router library, client library and router service
 
 [![Main CI](https://github.com/gammazero/nexus/actions/workflows/main-golint.yml/badge.svg)](https://github.com/gammazero/nexus/actions/workflows/main-golint.yml)
-[![Coverage](https://img.shields.io/badge/coverage-62.4%25-orange)](https://github.com/gammazero/nexus/actions/workflows/main-golint.yml)
+[![Coverage](https://img.shields.io/badge/coverage-62.6%25-orange)](https://github.com/gammazero/nexus/actions/workflows/main-golint.yml)
 [![License](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)
 [![GoDoc](https://godoc.org/github.com/gammazero/nexus?status.svg)](https://godoc.org/github.com/gammazero/nexus)
 
@@ -171,6 +171,7 @@ The currently maintained version of this module is 3.x. Earlier major versions a
 | registration_meta_procedures | Yes       |
 | pattern_based_registration   | Yes       |
 | shared_registration          | Yes       |
+| force_reregister             | Yes       |
 | sharded_registration         | No        |
 | registration_revocation      | No        |
 | procedure_reflection         | No        |

router/dealer.go

@@ -33,6 +33,7 @@ var dealerRole = wamp.Dict{ //nolint:gochecknoglobals
 		wamp.FeatureProgCallInvocations: true,
 		wamp.FeatureSessionMetaAPI:      true,
 		wamp.FeatureSharedReg:           true,
+		wamp.FeatureForceReregister:     true,
 		wamp.FeatureRegMetaAPI:          true,
 		wamp.FeatureTestamentMetaAPI:    true,
 		wamp.FeaturePayloadPassthruMode: true,
@@ -260,10 +261,11 @@ func (d *dealer) Register(callee *wamp.Session, msg *wamp.Register) {
 
 	invoke, _ := wamp.AsString(msg.Options[wamp.OptInvoke])
 	forwardTimeout, _ := msg.Options[wamp.OptForwardTimeout].(bool)
+	forceReregister, _ := msg.Options[wamp.OptForceReregister].(bool)
 	var metaPubs []*wamp.Publish
 	done := make(chan struct{})
 	d.actionChan <- func() {
-		metaPubs = d.syncRegister(callee, msg, match, invoke, disclose, forwardTimeout, wampURI)
+		metaPubs = d.syncRegister(callee, msg, match, invoke, disclose, forwardTimeout, forceReregister, wampURI)
 		close(done)
 	}
 	<-done
@@ -555,7 +557,7 @@ func (d *dealer) run() {
 	close(d.stopped)
 }
 
-func (d *dealer) syncRegister(callee *wamp.Session, msg *wamp.Register, match, invokePolicy string, disclose, forwardTimeout, wampURI bool) []*wamp.Publish { //nolint:lll
+func (d *dealer) syncRegister(callee *wamp.Session, msg *wamp.Register, match, invokePolicy string, disclose, forwardTimeout, forceReregister, wampURI bool) []*wamp.Publish { //nolint:lll
 	var metaPubs []*wamp.Publish
 	var reg *registration
 	switch match {
@@ -567,6 +569,15 @@ func (d *dealer) syncRegister(callee *wamp.Session, msg *wamp.Register, match, i
 		reg = d.wcProcRegMap[msg.Procedure]
 	}
 
+	// force_reregister lets a new callee forcibly take over a procedure
+	// currently held under invoke=single. Evict the prior callee(s), send
+	// them an unsolicited UNREGISTERED, fire meta events, then fall through
+	// to the fresh-registration branch below.
+	if reg != nil && forceReregister && (reg.policy == "" || reg.policy == wamp.InvokeSingle) {
+		metaPubs = append(metaPubs, d.syncEvictRegistration(reg, wampURI)...)
+		reg = nil
+	}
+
 	var created string
 	var regID wamp.ID
 	// If no existing registration found for the procedure, then create a new
@@ -1459,6 +1470,63 @@ func (d *dealer) syncRemoveSession(sess *wamp.Session) []*wamp.Publish {
 	return metaPubs
 }
 
+// syncEvictRegistration tears down an existing registration on behalf of a
+// force_reregister request. It sends an unsolicited UNREGISTERED message to
+// every callee currently attached to reg, drops reg from all dealer indexes
+// and per-callee reg sets, and returns the meta events that should be
+// published (one on_unregister per callee, plus a single on_delete).
+//
+// Callers run on the dealer actor goroutine.
+func (d *dealer) syncEvictRegistration(reg *registration, wampURI bool) []*wamp.Publish {
+	var metaPubs []*wamp.Publish
+	prevRegID := reg.id
+	for _, prev := range reg.callees {
+		d.trySend(prev, &wamp.Unregistered{
+			Details: wamp.Dict{
+				"registration": prevRegID,
+				"reason":       string(wamp.ErrUnregistered),
+			},
+		})
+		if set, ok := d.calleeRegIDSet[prev]; ok {
+			delete(set, prevRegID)
+			if len(set) == 0 {
+				delete(d.calleeRegIDSet, prev)
+			}
+		}
+		if !wampURI && d.metaPeer != nil {
+			metaPubs = append(metaPubs, &wamp.Publish{
+				Request:   wamp.GlobalID(),
+				Topic:     wamp.MetaEventRegOnUnregister,
+				Arguments: wamp.List{prev.ID, prevRegID},
+			})
+		}
+	}
+	delete(d.registrations, prevRegID)
+	switch reg.match {
+	default:
+		delete(d.procRegMap, reg.procedure)
+	case wamp.MatchPrefix:
+		delete(d.pfxProcRegMap, reg.procedure)
+	case wamp.MatchWildcard:
+		delete(d.wcProcRegMap, reg.procedure)
+	}
+	if !wampURI && d.metaPeer != nil && len(reg.callees) > 0 {
+		// on_delete uses the last callee's session ID, mirroring the order
+		// upstream uses elsewhere (see syncRemoveSession).
+		last := reg.callees[len(reg.callees)-1]
+		metaPubs = append(metaPubs, &wamp.Publish{
+			Request:   wamp.GlobalID(),
+			Topic:     wamp.MetaEventRegOnDelete,
+			Arguments: wamp.List{last.ID, prevRegID},
+		})
+	}
+	if d.debug {
+		d.log.Printf("Evicted registration %v for procedure %v (force_reregister)",
+			prevRegID, reg.procedure)
+	}
+	return metaPubs
+}
+
 // syncDelCalleeReg deletes the the callee from the specified registration and
 // deletes the registration from the set of registrations for the callee.
 //

router/dealer_test.go

@@ -1527,3 +1527,152 @@ func TestReceiveProgressForwardedWithoutCallCanceling(t *testing.T) {
 		"dealer should forward receive_progress when callee declares "+
 			"progressive_call_results, regardless of call_canceling")
 }
+
+// TestForceReregister pins the WAMP advanced-profile force_reregister
+// behavior: a second REGISTER for the same procedure with
+// force_reregister=true must evict the prior callee, send it an
+// unsolicited UNREGISTERED with Details.{registration,reason}, fire the
+// matching meta events, and install the new callee.
+func TestForceReregister(t *testing.T) {
+	dealer, metaClient := newTestDealer(t)
+
+	// Callee1 registers normally (default invoke=single).
+	callee1 := newTestPeer()
+	sess1 := wamp.NewSession(callee1, 0, nil, nil)
+	dealer.Register(sess1, &wamp.Register{Request: 1, Procedure: testProcedure})
+	regID1 := (<-callee1.Recv()).(*wamp.Registered).Registration
+	checkMetaReg(t, metaClient, sess1.ID) // on_create
+	checkMetaReg(t, metaClient, sess1.ID) // on_register
+
+	// Without force_reregister, a second REGISTER must fail.
+	callee2 := newTestPeer()
+	sess2 := wamp.NewSession(callee2, 0, nil, nil)
+	dealer.Register(sess2, &wamp.Register{Request: 2, Procedure: testProcedure})
+	errMsg, ok := (<-callee2.Recv()).(*wamp.Error)
+	require.True(t, ok, "expected ERROR without force_reregister")
+	require.Equal(t, wamp.ErrProcedureAlreadyExists, errMsg.Error)
+
+	// Now with force_reregister=true: callee2 should win.
+	dealer.Register(sess2, &wamp.Register{
+		Request:   3,
+		Procedure: testProcedure,
+		Options:   wamp.SetOption(nil, wamp.OptForceReregister, true),
+	})
+
+	// Callee1 receives an unsolicited UNREGISTERED carrying revocation details.
+	rev, ok := (<-callee1.Recv()).(*wamp.Unregistered)
+	require.True(t, ok, "evicted callee should receive UNREGISTERED")
+	require.Equal(t, wamp.ID(0), rev.Request, "revocation UNREGISTERED has Request=0")
+	require.NotNil(t, rev.Details, "revocation UNREGISTERED must carry Details")
+	revRegID, _ := wamp.AsID(rev.Details["registration"])
+	require.Equal(t, regID1, revRegID, "revocation Details.registration must match evicted regID")
+	require.Equal(t, string(wamp.ErrUnregistered), rev.Details["reason"])
+
+	// Callee2 receives a fresh REGISTERED.
+	reg2, ok := (<-callee2.Recv()).(*wamp.Registered)
+	require.True(t, ok, "callee2 should receive REGISTERED")
+	require.NotEqual(t, regID1, reg2.Registration,
+		"force_reregister should produce a new registration ID")
+
+	// Meta events: on_unregister for the evicted callee, on_delete for
+	// the torn-down registration, then on_create + on_register for the new
+	// registration installed for callee2.
+	checkMetaReg(t, metaClient, sess1.ID) // on_unregister (evicted)
+	checkMetaReg(t, metaClient, sess1.ID) // on_delete (last-seen sess id)
+	checkMetaReg(t, metaClient, sess2.ID) // on_create
+	checkMetaReg(t, metaClient, sess2.ID) // on_register
+
+	// Dealer state: only callee2 should own the procedure now.
+	reg, ok := dealer.procRegMap[testProcedure]
+	require.True(t, ok, "procedure registration missing after force_reregister")
+	require.Equal(t, reg2.Registration, reg.id)
+	require.Equal(t, 1, len(reg.callees))
+	require.Same(t, sess2, reg.callees[0])
+
+	// Callee1 must no longer appear in the dealer's reg-set bookkeeping.
+	_, stillHasCallee1 := dealer.calleeRegIDSet[sess1]
+	require.False(t, stillHasCallee1, "evicted callee should be removed from calleeRegIDSet")
+
+	// Calls should now invoke callee2.
+	caller := newTestPeer()
+	callerSess := wamp.NewSession(caller, 0, nil, nil)
+	dealer.Call(callerSess, &wamp.Call{Request: 99, Procedure: testProcedure})
+	select {
+	case msg := <-callee2.Recv():
+		_, ok := msg.(*wamp.Invocation)
+		require.True(t, ok, "expected INVOCATION on callee2")
+	case <-callee1.Recv():
+		require.FailNow(t, "evicted callee1 must not receive INVOCATION")
+	case <-time.After(time.Second):
+		require.FailNow(t, "timed out waiting for INVOCATION")
+	}
+}
+
+// TestForceReregisterIgnoredForSharedRegistration confirms force_reregister
+// does not tear down a registration whose invocation policy is shared
+// (anything other than single). The new caller should fall through to the
+// normal policy-conflict / shared-add path.
+func TestForceReregisterIgnoredForSharedRegistration(t *testing.T) {
+	dealer, metaClient := newTestDealer(t)
+
+	calleeRoles := wamp.Dict{
+		"roles": wamp.Dict{
+			"callee": wamp.Dict{
+				"features": wamp.Dict{
+					"shared_registration": true,
+				},
+			},
+		},
+	}
+
+	// Callee1 holds a roundrobin shared registration.
+	callee1 := newTestPeer()
+	sess1 := wamp.NewSession(callee1, 0, nil, calleeRoles)
+	dealer.Register(sess1, &wamp.Register{
+		Request:   1,
+		Procedure: testProcedure,
+		Options:   wamp.SetOption(nil, wamp.OptInvoke, wamp.InvokeRoundRobin),
+	})
+	regID1 := (<-callee1.Recv()).(*wamp.Registered).Registration
+	checkMetaReg(t, metaClient, sess1.ID)
+	checkMetaReg(t, metaClient, sess1.ID)
+
+	// Callee2 requests force_reregister=true but invoke=single — the
+	// existing registration is shared, so eviction must be skipped and
+	// the policy-conflict error must fire.
+	callee2 := newTestPeer()
+	sess2 := wamp.NewSession(callee2, 0, nil, calleeRoles)
+	opts := wamp.SetOption(nil, wamp.OptForceReregister, true)
+	opts = wamp.SetOption(opts, wamp.OptInvoke, wamp.InvokeSingle)
+	dealer.Register(sess2, &wamp.Register{
+		Request:   2,
+		Procedure: testProcedure,
+		Options:   opts,
+	})
+	errMsg, ok := (<-callee2.Recv()).(*wamp.Error)
+	require.True(t, ok, "expected ERROR — force_reregister must not evict shared regs")
+	require.Equal(t, wamp.ErrProcedureAlreadyExists, errMsg.Error)
+
+	// Original registration must still exist with callee1 attached.
+	reg, ok := dealer.registrations[regID1]
+	require.True(t, ok, "original shared registration was unexpectedly removed")
+	require.Equal(t, 1, len(reg.callees))
+	require.Same(t, sess1, reg.callees[0])
+
+	// Callee1 should not have received any UNREGISTERED.
+	select {
+	case msg := <-callee1.Recv():
+		require.FailNow(t, "callee1 should not receive any message",
+			"got %T", msg)
+	default:
+	}
+}
+
+// TestForceReregisterAdvertisedFeature ensures the dealer announces
+// force_reregister in its role features so clients can discover support.
+func TestForceReregisterAdvertisedFeature(t *testing.T) {
+	features, ok := dealerRole["features"].(wamp.Dict)
+	require.True(t, ok, "dealer role missing features dict")
+	supported, _ := features[wamp.FeatureForceReregister].(bool)
+	require.True(t, supported, "dealer must advertise force_reregister=true")
+}

test/spec_force_reregister_test.go

@@ -0,0 +1,64 @@
+package test_test
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/gammazero/nexus/v3/client"
+	"github.com/gammazero/nexus/v3/wamp"
+)
+
+// TestSpecForceReregister exercises the WAMP advanced-profile
+// force_reregister option end-to-end across whichever transport the
+// test matrix is currently running on. The second callee must forcibly
+// take over the procedure from the first, and subsequent CALLs must be
+// routed to the new callee.
+func TestSpecForceReregister(t *testing.T) {
+	checkGoLeaks(t)
+
+	const procName = "nexus.test.force_reregister"
+
+	callee1 := connectClient(t)
+	callee2 := connectClient(t)
+	caller := connectClient(t)
+
+	var hits1, hits2 atomic.Int32
+	handler1 := func(_ context.Context, _ *wamp.Invocation) client.InvokeResult {
+		hits1.Add(1)
+		return client.InvokeResult{Args: wamp.List{"callee1"}}
+	}
+	handler2 := func(_ context.Context, _ *wamp.Invocation) client.InvokeResult {
+		hits2.Add(1)
+		return client.InvokeResult{Args: wamp.List{"callee2"}}
+	}
+
+	require.NoError(t, callee1.Register(procName, handler1, nil),
+		"callee1 should be able to register the procedure")
+
+	// Second REGISTER without force_reregister must fail with
+	// procedure_already_exists.
+	err := callee2.Register(procName, handler2, nil)
+	require.Error(t, err,
+		"second register without force_reregister must fail")
+
+	// Now callee2 force-reregisters; should succeed and evict callee1.
+	require.NoError(t,
+		callee2.Register(procName, handler2,
+			wamp.Dict{wamp.OptForceReregister: true}),
+		"force_reregister must let callee2 take over")
+
+	// A CALL must now be routed to callee2.
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	res, err := caller.Call(ctx, procName, nil, nil, nil, nil)
+	require.NoError(t, err, "post-force_reregister CALL should succeed")
+	require.NotEmpty(t, res.Arguments)
+	require.Equal(t, "callee2", res.Arguments[0],
+		"force_reregister should route subsequent CALLs to the new callee")
+	require.Equal(t, int32(1), hits2.Load(), "callee2 should have served the call")
+	require.Equal(t, int32(0), hits1.Load(), "callee1 must not have served the call")
+}

test/spec_unimplemented_test.go

@@ -36,10 +36,6 @@ func TestSpecUnimplementedTopicReflection(t *testing.T) {
 	t.Skip("pending: topic_reflection — wamp.reflection.topic.* not implemented")
 }
 
-func TestSpecUnimplementedForceReregister(t *testing.T) {
-	t.Skip("pending: force_reregister — option not honored by dealer.Register")
-}
-
 func TestSpecUnimplementedBatchedWSTransport(t *testing.T) {
 	t.Skip("pending: batched WebSocket transport (wamp.2.json.batched / msgpack.batched)")
 }

wamp/message.go

@@ -335,8 +335,15 @@ func (msg *Unregister) MessageType() MessageType { return UNREGISTER }
 // the Callee:
 //
 // [UNREGISTERED, UNREGISTER.Request|id]
+//
+// The Dealer also sends UNREGISTERED unsolicited when a registration is
+// revoked (e.g. force_reregister); the extended form carries a Details dict
+// with "registration" and "reason":
+//
+// [UNREGISTERED, 0, Details|dict]
 type Unregistered struct {
 	Request ID
+	Details Dict `wamp:"omitempty"`
 }
 
 func (msg *Unregistered) MessageType() MessageType { return UNREGISTERED }

wamp/options.go

@@ -23,6 +23,7 @@ const (
 	OptPPTKeyId        = "ppt_keyid"
 	OptSticky          = "sticky"
 	OptForwardTimeout  = "forward_timeout"
+	OptForceReregister = "force_reregister"
 
 	// Values for URI matching mode.
 	MatchExact    = "exact"

wamp/roles_reatures.go

@@ -18,6 +18,7 @@ const (
 	FeatureProgCallInvocations = "progressive_call_invocations"
 	FeatureSessionMetaAPI      = "session_meta_api"
 	FeatureSharedReg           = "shared_registration"
+	FeatureForceReregister     = "force_reregister"
 	FeatureRegMetaAPI          = "registration_meta_api"
 	FeatureTestamentMetaAPI    = "testament_meta_api"
 

wamp/uris.go

@@ -22,6 +22,11 @@ const (
 	// is not active.
 	ErrNoSuchRegistration = URI("wamp.error.no_such_registration")
 
+	// Sent by a Dealer in an unsolicited UNREGISTERED message to a Callee
+	// whose registration has been revoked — e.g. because another peer forcibly
+	// took it over with force_reregister=true.
+	ErrUnregistered = URI("wamp.error.unregistered")
+
 	// A Broker could not perform an unsubscribe, since the given subscription
 	// is not active.
 	ErrNoSuchSubscription = URI("wamp.error.no_such_subscription")