state: refresh the NodeStore copies of a user in UpdateUser

hscontrol/servertest/issues_test.go

@@ -1239,3 +1239,34 @@ func peerByHostname(nm *netmap.NetworkMap, hostname string) (tailcfg.NodeView, b
 
 	return tailcfg.NodeView{}, false
 }
+
+func TestUserProfileFollowsRename(t *testing.T) {
+	srv := servertest.NewServer(t)
+	owner := srv.CreateUser(t, "profile-original")
+	profileID := owner.TailscaleUserProfile().ID
+	peer := srv.CreateUser(t, "profile-peer")
+	ownerClient := servertest.NewClient(t, srv, "owner", servertest.WithUser(owner))
+	peerClient := servertest.NewClient(t, srv, "peer", servertest.WithUser(peer))
+	ownerClient.WaitForPeers(t, 1, 5*time.Second)
+	peerClient.WaitForPeers(t, 1, 5*time.Second)
+
+	_, c, err := srv.State().RenameUser(types.UserID(owner.ID), "profile-renamed")
+	require.NoError(t, err)
+
+	srv.App.Change(c)
+
+	for _, client := range []*servertest.TestClient{ownerClient, peerClient} {
+		client.WaitForCondition(t, "renamed user profile", 5*time.Second,
+			func(nm *netmap.NetworkMap) bool {
+				profile, ok := nm.UserProfiles[profileID]
+				return ok && profile.LoginName() == "profile-renamed"
+			})
+	}
+
+	late := servertest.NewClient(t, srv, "late", servertest.WithUser(peer))
+	late.WaitForCondition(t, "renamed user profile on join", 5*time.Second,
+		func(nm *netmap.NetworkMap) bool {
+			profile, ok := nm.UserProfiles[profileID]
+			return ok && profile.LoginName() == "profile-renamed"
+		})
+}

hscontrol/state/persist_test.go

@@ -2350,3 +2350,264 @@ func TestUpdateUserClearsColumns(t *testing.T) {
 	assert.Empty(t, user.ProfilePicURL)
 	assert.Equal(t, "renamed", user.Name)
 }
+
+func TestUpdateUserRefreshesNodeCopies(t *testing.T) {
+	_, s, nodeID := persistTestSetup(t)
+	t.Cleanup(func() { _ = s.Close() })
+
+	original, ok := s.GetNodeByID(nodeID)
+	require.True(t, ok)
+
+	userID := types.UserID(original.UserID().Get())
+	user, err := s.GetUserByID(userID)
+	require.NoError(t, err)
+
+	_, ok = s.nodeStore.UpdateNode(nodeID, func(n *types.Node) {
+		n.AuthKey = &types.Credential{ID: 1, UserID: new(user.ID), User: user.Clone()}
+	})
+	require.True(t, ok)
+
+	tagged := s.nodeStore.PutNode(types.Node{
+		ID: nodeID + 1, Hostname: "tagged", Tags: []string{"tag:server"},
+		MachineKey: key.NewMachine().Public(), NodeKey: key.NewNode().Public(),
+		AuthKey: &types.Credential{ID: 2, UserID: new(user.ID), User: user.Clone()},
+	})
+	other, _, err := s.CreateUser(types.User{Name: "other"})
+	require.NoError(t, err)
+
+	otherNode := s.nodeStore.PutNode(types.Node{
+		ID: nodeID + 2, Hostname: "other", UserID: new(other.ID), User: other.Clone(),
+		MachineKey: key.NewMachine().Public(), NodeKey: key.NewNode().Public(),
+		AuthKey: &types.Credential{ID: 3, UserID: new(other.ID), User: other.Clone()},
+	})
+	before, ok := s.GetNodeByID(nodeID)
+	require.True(t, ok)
+
+	_, c, err := s.RenameUser(userID, "renamed-user")
+	require.NoError(t, err)
+	assert.True(t, c.IncludeSelf)
+	assert.True(t, c.SendAllPeers)
+
+	updated, ok := s.GetNodeByID(nodeID)
+	require.True(t, ok)
+	assert.Equal(t, "renamed-user", updated.User().Name())
+	assert.Equal(t, "renamed-user", updated.AuthKey().User().Name())
+	assert.Equal(t, "persist-user", before.User().Name())
+	assert.Equal(t, "persist-user", before.AuthKey().User().Name())
+
+	updatedTagged, ok := s.GetNodeByID(tagged.ID())
+	require.True(t, ok)
+	assert.True(t, updatedTagged.IsTagged())
+	assert.Equal(t, "renamed-user", updatedTagged.AuthKey().User().Name())
+	assert.Equal(t, "persist-user", tagged.AuthKey().User().Name())
+
+	unchanged, ok := s.GetNodeByID(otherNode.ID())
+	require.True(t, ok)
+	assert.Equal(t, "other", unchanged.User().Name())
+	assert.Equal(t, "other", unchanged.AuthKey().User().Name())
+
+	snapshot := s.nodeStore.data.Load()
+	_, c, err = s.RenameUser(userID, "renamed-user")
+	require.NoError(t, err)
+	assert.True(t, c.IsEmpty())
+	assert.Same(t, snapshot, s.nodeStore.data.Load(), "an unchanged user must not rebuild the snapshot")
+
+	_, c, err = s.UpdateUser(userID, func(user *types.User) error {
+		user.Provider = "updated-provider"
+		return nil
+	})
+	require.NoError(t, err)
+	assert.True(t, c.IsEmpty(), "a non-profile column does not need a full map update")
+
+	updated, ok = s.GetNodeByID(nodeID)
+	require.True(t, ok)
+	assert.Equal(t, "updated-provider", updated.User().Provider())
+	assert.Equal(t, "updated-provider", updated.AuthKey().User().Provider())
+}
+
+type blockingSetUsersPolicyManager struct {
+	policy.PolicyManager
+
+	blocked atomic.Bool
+	reached chan struct{}
+	resume  <-chan struct{}
+	last    []types.User
+}
+
+func (pm *blockingSetUsersPolicyManager) SetUsers(users []types.User) (bool, bool, error) {
+	if !pm.blocked.Swap(true) {
+		close(pm.reached)
+		<-pm.resume
+	}
+
+	pm.last = users
+
+	return pm.PolicyManager.SetUsers(users)
+}
+
+func TestUpdateUserPublishesInOrder(t *testing.T) {
+	_, s, nodeID := persistTestSetup(t)
+	t.Cleanup(func() { _ = s.Close() })
+
+	node, ok := s.GetNodeByID(nodeID)
+	require.True(t, ok)
+
+	user := node.User().AsStruct()
+	userID := types.UserID(user.ID)
+	_, ok = s.nodeStore.UpdateNode(nodeID, func(n *types.Node) {
+		n.AuthKey = &types.Credential{ID: 1, UserID: new(user.ID), User: user.Clone()}
+	})
+	require.True(t, ok)
+
+	resume := make(chan struct{})
+
+	releaseFirst := sync.OnceFunc(func() { close(resume) })
+	defer releaseFirst()
+
+	pm := &blockingSetUsersPolicyManager{
+		PolicyManager: s.polMan,
+		reached:       make(chan struct{}),
+		resume:        resume,
+	}
+	s.polMan = pm
+
+	update := func(name string) error {
+		_, _, err := s.UpdateUser(userID, func(u *types.User) error {
+			u.Name = name
+			u.Email = name + "@example.com"
+			u.DisplayName = name + " display"
+			u.ProfilePicURL = "https://example.com/" + name
+
+			return nil
+		})
+
+		return err
+	}
+
+	firstDone := make(chan error, 1)
+	go func() { firstDone <- update("first-user") }()
+
+	// The first update pauses after its commit, inside the policy user update.
+	<-pm.reached
+
+	var secondErr error
+
+	secondDone := make(chan struct{})
+
+	go func() {
+		secondErr = update("latest-user")
+
+		close(secondDone)
+	}()
+
+	// Give the second update time to overtake the paused first one.
+	select {
+	case <-secondDone:
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	releaseFirst()
+	require.NoError(t, <-firstDone)
+	<-secondDone
+	require.NoError(t, secondErr)
+
+	stored, err := s.GetUserByID(userID)
+	require.NoError(t, err)
+	assert.Equal(t, "latest-user", stored.Name)
+
+	require.Len(t, pm.last, 1)
+	assert.Equal(t, stored.Name, pm.last[0].Name, "policy users must follow the database")
+
+	updated, ok := s.GetNodeByID(nodeID)
+	require.True(t, ok)
+	assert.Equal(t, stored.Name, updated.User().Name())
+	assert.Equal(t, stored.Name, updated.AuthKey().User().Name())
+	assert.Equal(t, stored.TailscaleUserProfile(), updated.User().TailscaleUserProfile())
+	assert.Equal(t, stored.TailscaleUserProfile(), updated.AuthKey().User().TailscaleUserProfile())
+}
+
+func TestUserListPublishesInOrder(t *testing.T) {
+	_, s, nodeID := persistTestSetup(t)
+	t.Cleanup(func() { _ = s.Close() })
+
+	node, ok := s.GetNodeByID(nodeID)
+	require.True(t, ok)
+
+	resume := make(chan struct{})
+
+	releaseFirst := sync.OnceFunc(func() { close(resume) })
+	defer releaseFirst()
+
+	pm := &blockingSetUsersPolicyManager{
+		PolicyManager: s.polMan,
+		reached:       make(chan struct{}),
+		resume:        resume,
+	}
+	s.polMan = pm
+
+	createDone := make(chan error, 1)
+
+	go func() {
+		_, _, err := s.CreateUser(types.User{Name: "created-user"})
+		createDone <- err
+	}()
+
+	// The creation pauses holding a user list read before the rename commits.
+	<-pm.reached
+
+	var renameErr error
+
+	renameDone := make(chan struct{})
+
+	go func() {
+		_, _, renameErr = s.RenameUser(types.UserID(node.User().ID()), "renamed-user")
+
+		close(renameDone)
+	}()
+
+	select {
+	case <-renameDone:
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	releaseFirst()
+	require.NoError(t, <-createDone)
+	<-renameDone
+	require.NoError(t, renameErr)
+
+	names := make([]string, 0, len(pm.last))
+	for _, user := range pm.last {
+		names = append(names, user.Name)
+	}
+
+	assert.ElementsMatch(t, []string{"renamed-user", "created-user"}, names,
+		"policy users must follow the database")
+}
+
+var errSetUsersTest = errors.New("set users failed")
+
+type failingSetUsersPolicyManager struct {
+	policy.PolicyManager
+}
+
+func (failingSetUsersPolicyManager) SetUsers([]types.User) (bool, bool, error) {
+	return false, false, errSetUsersTest
+}
+
+func TestUpdateUserRefreshesNodeCopiesWhenPolicyUpdateFails(t *testing.T) {
+	_, s, nodeID := persistTestSetup(t)
+	t.Cleanup(func() { _ = s.Close() })
+
+	node, ok := s.GetNodeByID(nodeID)
+	require.True(t, ok)
+
+	s.polMan = failingSetUsersPolicyManager{PolicyManager: s.polMan}
+
+	_, _, err := s.RenameUser(types.UserID(node.User().ID()), "renamed-user")
+	require.ErrorIs(t, err, errSetUsersTest)
+
+	updated, ok := s.GetNodeByID(nodeID)
+	require.True(t, ok)
+	assert.Equal(t, "renamed-user", updated.User().Name(),
+		"the committed row and its cached copy must agree")
+}

hscontrol/state/state.go

@@ -227,6 +227,13 @@ type State struct {
 	// nothing.
 	persistMu sync.Mutex
 
+	// userUpdateMu keeps committed user changes and cached copies in the same order.
+	userUpdateMu sync.Mutex
+
+	// policyUsersMu orders user-list reads with SetUsers, so a list read
+	// before a commit cannot replace one read after it.
+	policyUsersMu sync.Mutex
+
 	// selfRefresh holds nodes whose own DNS config inputs changed (NextDNS
 	// device metadata, given name), drained by [State.DrainSelfRefreshes].
 	selfRefresh   []types.NodeID
@@ -528,12 +535,19 @@ func (s *State) CreateUser(user types.User) (*types.User, change.Change, error)
 // UpdateUser modifies an existing user using the provided update function within a transaction.
 // Returns the updated user, change set, and any error.
 func (s *State) UpdateUser(userID types.UserID, updateFn func(*types.User) error) (*types.User, change.Change, error) {
+	s.userUpdateMu.Lock()
+	defer s.userUpdateMu.Unlock()
+
+	var previous types.User
+
 	user, err := hsdb.Write(s.db.DB, func(tx *gorm.DB) (*types.User, error) {
 		user, err := hsdb.GetUserByID(tx, userID)
 		if err != nil {
 			return nil, err
 		}
 
+		previous = *user
+
 		if err := updateFn(user); err != nil { //nolint:noinlineerr
 			return nil, err
 		}
@@ -551,13 +565,57 @@ func (s *State) UpdateUser(userID types.UserID, updateFn func(*types.User) error
 		return nil, change.Change{}, err
 	}
 
+	var profileChanged bool
+
+	previous.UpdatedAt = user.UpdatedAt
+	if previous != *user {
+		updates := make(map[types.NodeID]UpdateNodeFunc)
+
+		var ownerRefreshed bool
+
+		for _, node := range s.nodeStore.ListNodes().All() {
+			owner := node.User()
+			credential := node.AuthKey()
+			owns := owner.Valid() && owner.ID() == user.ID
+			createdKey := credential.Valid() && credential.User().Valid() &&
+				credential.User().ID() == user.ID
+
+			if !owns && !createdKey {
+				continue
+			}
+
+			updates[node.ID()] = func(n *types.Node) {
+				// Recheck ownership after waiting in the writer queue, and replace
+				// shared records so earlier snapshots keep their user data.
+				if n.User != nil && n.User.ID == user.ID {
+					n.User = user.Clone()
+					ownerRefreshed = ownerRefreshed || !n.IsTagged()
+				}
+
+				if n.AuthKey != nil && n.AuthKey.User != nil && n.AuthKey.User.ID == user.ID {
+					n.AuthKey = n.AuthKey.Clone()
+					n.AuthKey.User = user.Clone()
+				}
+			}
+		}
+
+		s.nodeStore.UpdateNodes(updates)
+
+		previousProfile := previous.TailscaleUserProfile()
+		profile := user.TailscaleUserProfile()
+
+		profileChanged = ownerRefreshed && !previousProfile.Equal(&profile)
+	}
+
 	// Check if policy manager needs updating
 	c, err := s.updatePolicyManagerUsers()
 	if err != nil {
 		return user, change.Change{}, fmt.Errorf("updating policy manager after user update: %w", err)
 	}
 
-	// TODO(kradalby): We might want to update nodestore with the user data
+	if profileChanged {
+		c = change.UserAdded()
+	}
 
 	return user, c, nil
 }
@@ -3726,6 +3784,9 @@ func reloginChanged(before, after types.NodeView, now time.Time) relogin {
 // manager, rebuilds peer adjacency when user identity changed, and returns
 // a PolicyChange when clients need a refresh.
 func (s *State) updatePolicyManagerUsers() (change.Change, error) {
+	s.policyUsersMu.Lock()
+	defer s.policyUsersMu.Unlock()
+
 	users, err := s.ListAllUsers()
 	if err != nil {
 		return change.Change{}, fmt.Errorf("listing users for policy update: %w", err)