services/syncbase/discovery/discovery.go - release.go.x.ref - Git at Google

 // Copyright 2016 The Vanadium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.

 package discovery

 import (
 	"bytes"
 	"fmt"
 	"sort"
 	"strings"
 	"time"

 	"v.io/v23"
 	"v.io/v23/context"
 	"v.io/v23/discovery"
 	"v.io/v23/security"
 	wire "v.io/v23/services/syncbase"
 	"v.io/x/lib/nsync"
 	"v.io/x/ref/lib/discovery/global"
 	"v.io/x/ref/services/syncbase/server/interfaces"
 )

 const (
 	visibilityKey = "vis"

 	nhDiscoveryKey = iota
 	globalDiscoveryKey
 )

 // Discovery implements v.io/v23/discovery.T for syncbase based
 // applications.
 // TODO(mattr): Actually this is not syncbase specific.  At some
 // point we should just replace the result of v23.NewDiscovery
 // with this.
 type Discovery struct {
 	nhDiscovery     discovery.T
 	globalDiscovery discovery.T
 }

 // NewDiscovery creates a new syncbase discovery object.
 // globalDiscoveryPath is the path in the namespace where global disovery
 // advertisements will be mounted.
 // If globalDiscoveryPath is empty, no global discovery service will be created.
 // globalScanInterval is the interval at which global discovery will be refreshed.
 // If globalScanInterval is 0, the defaultScanInterval of global discovery will
 // be used.
 func NewDiscovery(ctx *context.T, globalDiscoveryPath string, globalScanInterval time.Duration) (discovery.T, error) {
 	d := &Discovery{}
 	var err error
 	if d.nhDiscovery, err = v23.NewDiscovery(ctx); err != nil {
 		return nil, err
 	}
 	if globalDiscoveryPath != "" {
 		if d.globalDiscovery, err = global.NewWithTTL(ctx, globalDiscoveryPath, 0, globalScanInterval); err != nil {
 			return nil, err
 		}
 	}
 	return d, nil
 }

 // Scan implements v.io/v23/discovery/T.Scan.
 func (d *Discovery) Scan(ctx *context.T, query string) (<-chan discovery.Update, error) {
 	nhCtx, nhCancel := context.WithCancel(ctx)
 	nhUpdates, err := d.nhDiscovery.Scan(nhCtx, query)
 	if err != nil {
 		nhCancel()
 		return nil, err
 	}
 	var globalUpdates <-chan discovery.Update
 	if d.globalDiscovery != nil {
 		if globalUpdates, err = d.globalDiscovery.Scan(ctx, query); err != nil {
 			nhCancel()
 			return nil, err
 		}
 	}

 	// Currently setting visibility on the neighborhood discovery
 	// service turns IBE encryption on.  We currently don't have the
 	// infrastructure support for IBE, so that would make our advertisements
 	// unreadable by everyone.
 	// Instead we add the visibility list to the attributes of the advertisement
 	// and filter on the client side.  This is a temporary measure until
 	// IBE is set up.  See v.io/i/1345.
 	updates := make(chan discovery.Update)
 	go func() {
 		defer close(updates)
 		seen := make(map[discovery.AdId]*updateRef)
 		for {
 			var u discovery.Update
 			var src uint // key of the source discovery service where the update came from
 			select {
 			case <-ctx.Done():
 				return
 			case u = <-nhUpdates:
 				src = nhDiscoveryKey
 			case u = <-globalUpdates:
 				src = globalDiscoveryKey
 			}
 			d.handleUpdate(ctx, u, src, seen, updates)
 		}
 	}()

 	return updates, nil
 }

 func (d *Discovery) handleUpdate(ctx *context.T, u discovery.Update, src uint, seen map[discovery.AdId]*updateRef, updates chan discovery.Update) {
 	if u == nil {
 		return
 	}
 	patterns := splitPatterns(u.Attribute(visibilityKey))
 	if len(patterns) > 0 && !matchesPatterns(ctx, patterns) {
 		return
 	}

 	id := u.Id()
 	prev := seen[id]
 	if u.IsLost() {
 		// Only send the lost noitification if a found event was previously seen,
 		// and all discovery services that found it have lost it.
 		if prev == nil || !prev.unset(src) {
 			return
 		}
 		delete(seen, id)
 		updates <- update{Update: u, lost: true}
 		return
 	}

 	if prev == nil {
 		// Always send updates for updates that we have never seen before.
 		ref := &updateRef{update: u}
 		ref.set(src)
 		seen[id] = ref
 		updates <- update{Update: u}
 		return
 	}

 	if differ := updatesDiffer(prev.update, u); (differ && u.Timestamp().After(prev.update.Timestamp())) ||
 		(!differ && src == nhDiscoveryKey && len(u.Advertisement().Attachments) > 0) {
 		// If the updates differ and the newly found update has a later time than
 		// previously found one, lose prev and find new.
 		// Or, if the update doesn't differ, but is from neighborhood discovery, it
 		// could have more information since we don't yet encode attachements in
 		// global discovery.
 		updates <- update{Update: prev.update, lost: true}
 		ref := &updateRef{update: u}
 		ref.set(src)
 		seen[id] = ref
 		updates <- update{Update: u}
 		return
 	}
 }

 // Advertise implements v.io/v23/discovery/T.Advertise.
 func (d *Discovery) Advertise(ctx *context.T, ad *discovery.Advertisement, visibility []security.BlessingPattern) (<-chan struct{}, error) {
 	// Currently setting visibility on the neighborhood discovery
 	// service turns IBE encryption on.  We currently don't have the
 	// infrastructure support for IBE, so that would make our advertisements
 	// unreadable by everyone.
 	// Instead we add the visibility list to the attributes of the advertisement
 	// and filter on the client side.  This is a temporary measure until
 	// IBE is set up.  See v.io/i/1345.
 	adCopy := *ad
 	if len(visibility) > 0 {
 		adCopy.Attributes = make(discovery.Attributes, len(ad.Attributes)+1)
 		for k, v := range ad.Attributes {
 			adCopy.Attributes[k] = v
 		}
 		patterns := joinPatterns(visibility)
 		adCopy.Attributes[visibilityKey] = patterns
 	}

 	stopped := make(chan struct{})
 	nhCtx, nhCancel := context.WithCancel(ctx)
 	nhStopped, err := d.nhDiscovery.Advertise(nhCtx, &adCopy, nil)
 	if err != nil {
 		nhCancel()
 		return nil, err
 	}
 	var globalStopped <-chan struct{}
 	if d.globalDiscovery != nil {
 		if globalStopped, err = d.globalDiscovery.Advertise(ctx, &adCopy, nil); err != nil {
 			nhCancel()
 			<-nhStopped
 			return nil, err
 		}
 	}
 	go func() {
 		<-nhStopped
 		if d.globalDiscovery != nil {
 			<-globalStopped
 		}
 		close(stopped)
 	}()
 	ad.Id = adCopy.Id
 	return stopped, nil
 }

 func updatesDiffer(a, b discovery.Update) bool {
 	if !sortedStringsEqual(a.Addresses(), b.Addresses()) {
 		return true
 	}
 	if !mapsEqual(a.Advertisement().Attributes, b.Advertisement().Attributes) {
 		return true
 	}
 	return false
 }

 func mapsEqual(a, b map[string]string) bool {
 	if len(a) != len(b) {
 		return false
 	}
 	for ka, va := range a {
 		if vb, ok := b[ka]; !ok || va != vb {
 			return false
 		}
 	}
 	return true
 }

 func sortedStringsEqual(a, b []string) bool {
 	// We want to make a nil and an empty slices equal to avoid unnecessary inequality by that.
 	if len(a) != len(b) {
 		return false
 	}
 	for i, v := range a {
 		if v != b[i] {
 			return false
 		}
 	}
 	return true
 }

 func matchesPatterns(ctx *context.T, patterns []security.BlessingPattern) bool {
 	p := v23.GetPrincipal(ctx)
 	blessings := p.BlessingStore().PeerBlessings()
 	for _, b := range blessings {
 		names := security.BlessingNames(p, b)
 		for _, pattern := range patterns {
 			if pattern.MatchedBy(names...) {
 				return true
 			}
 		}
 	}
 	return false
 }

 type updateRef struct {
 	update    discovery.Update
 	nhRef     bool
 	globalRef bool
 }

 func (r *updateRef) set(d uint) {
 	switch d {
 	case nhDiscoveryKey:
 		r.nhRef = true
 	case globalDiscoveryKey:
 		r.globalRef = true
 	}
 }

 func (r *updateRef) unset(d uint) bool {
 	switch d {
 	case nhDiscoveryKey:
 		r.nhRef = false
 	case globalDiscoveryKey:
 		r.globalRef = false
 	}
 	return !r.nhRef && !r.globalRef
 }

 // update wraps the discovery.Update to remove the visibility attribute which we add
 // and allows us to mark the update as lost.
 type update struct {
 	discovery.Update
 	lost bool
 }

 func (u update) IsLost() bool { return u.lost }

 func (u update) Attribute(name string) string {
 	if name == visibilityKey {
 		return ""
 	}
 	return u.Update.Attribute(name)
 }

 func (u update) Advertisement() discovery.Advertisement {
 	cp := u.Update.Advertisement()
 	orig := cp.Attributes
 	cp.Attributes = make(discovery.Attributes, len(orig))
 	for k, v := range orig {
 		if k != visibilityKey {
 			cp.Attributes[k] = v
 		}
 	}
 	return cp
 }

 // blessingSeparator is used to join multiple blessings into a
 // single string.
 // Note that comma cannot appear in blessings, see:
 // v.io/v23/security/certificate.go
 const blessingsSeparator = ','

 // joinPatterns concatenates the elements of a to create a single string.
 // The string can be split again with SplitPatterns.
 func joinPatterns(a []security.BlessingPattern) string {
 	if len(a) == 0 {
 		return ""
 	}
 	if len(a) == 1 {
 		return string(a[0])
 	}
 	n := (len(a) - 1)
 	for i := 0; i < len(a); i++ {
 		n += len(a[i])
 	}

 	b := make([]byte, n)
 	bp := copy(b, a[0])
 	for _, s := range a[1:] {
 		b[bp] = blessingsSeparator
 		bp++
 		bp += copy(b[bp:], s)
 	}
 	return string(b)
 }

 // splitPatterns splits BlessingPatterns that were joined with
 // JoinBlessingPattern.
 func splitPatterns(patterns string) []security.BlessingPattern {
 	if patterns == "" {
 		return nil
 	}
 	n := strings.Count(patterns, string(blessingsSeparator)) + 1
 	out := make([]security.BlessingPattern, n)
 	last, start := 0, 0
 	for i, r := range patterns {
 		if r == blessingsSeparator {
 			out[last] = security.BlessingPattern(patterns[start:i])
 			last++
 			start = i + 1
 		}
 	}
 	out[last] = security.BlessingPattern(patterns[start:])
 	return out
 }

 var state struct {
 	mu    nsync.Mu
 	scans map[security.Principal]*scanState
 }

 type scanState struct {
 	dbs       map[wire.Id]*inviteQueue
 	listeners int
 	cancel    context.CancelFunc
 }

 func newScan(ctx *context.T) (*scanState, error) {
 	ctx, cancel := context.WithRootCancel(ctx)
 	scan := &scanState{
 		dbs:    make(map[wire.Id]*inviteQueue),
 		cancel: cancel,
 	}
 	// TODO(suharshs): Add globalDiscoveryPath.
 	d, err := NewDiscovery(ctx, "", 0)
 	if err != nil {
 		scan.cancel()
 		return nil, err
 	}
 	query := fmt.Sprintf("v.InterfaceName=\"%s/%s\"",
 		interfaces.SyncDesc.PkgPath, interfaces.SyncDesc.Name)
 	updates, err := d.Scan(ctx, query)
 	if err != nil {
 		scan.cancel()
 		return nil, err
 	}
 	go func() {
 		for u := range updates {
 			if invite, db, ok := makeInvite(u.Advertisement()); ok {
 				state.mu.Lock()
 				q := scan.dbs[db]
 				if u.IsLost() {
 					// TODO(mattr): Removing like this can result in resurfacing already
 					// retrieved invites.  For example if the ACL of a syncgroup is
 					// changed to add a new memeber, then we might see a remove and then
 					// later an add.  In this case we would end up moving the invite
 					// to the end of the queue.  One way to fix this would be to
 					// keep removed invites for some time.
 					if q != nil {
 						q.remove(invite)
 						if q.empty() {
 							delete(scan.dbs, db)
 						}
 					}
 				} else {
 					if q == nil {
 						q = newInviteQueue()
 						scan.dbs[db] = q
 					}
 					q.add(invite)
 					q.cond.Broadcast()
 				}
 				state.mu.Unlock()
 			}
 		}
 	}()
 	return scan, nil
 }

 // Invite represents an invitation to join a syncgroup as found via Discovery.
 type Invite struct {
 	Syncgroup wire.Id  //Syncgroup is the Id of the syncgroup you've been invited to.
 	Addresses []string //Addresses are the list of addresses of the inviting server.
 	key       string
 }

 // ListenForInvites listens via Discovery for syncgroup invitations for the given
 // database and sends the invites to the provided channel.  We stop listening when
 // the given context is canceled.  When that happens we close the given channel.
 func ListenForInvites(ctx *context.T, db wire.Id, ch chan<- Invite) error {
 	defer state.mu.Unlock()
 	state.mu.Lock()

 	p := v23.GetPrincipal(ctx)
 	scan := state.scans[p]
 	if scan == nil {
 		var err error
 		if scan, err = newScan(ctx); err != nil {
 			return err
 		}
 		if state.scans == nil {
 			state.scans = make(map[security.Principal]*scanState)
 		}
 		state.scans[p] = scan
 	}
 	scan.listeners++

 	q := scan.dbs[db]
 	if q == nil {
 		q = newInviteQueue()
 		scan.dbs[db] = q
 	}

 	go func() {
 		c := q.scan()
 		for {
 			next, ok := q.next(ctx, c)
 			if !ok {
 				break
 			}
 			select {
 			case ch <- next.copy():
 			case <-ctx.Done():
 				break
 			}
 		}
 		close(ch)

 		defer state.mu.Unlock()
 		state.mu.Lock()

 		if q.empty() {
 			delete(scan.dbs, db)
 		}
 		scan.listeners--
 		if scan.listeners == 0 {
 			scan.cancel()
 			delete(state.scans, v23.GetPrincipal(ctx))
 		}
 	}()

 	return nil
 }

 // inviteQueue is a linked list based queue.  As we get new invitations we
 // add them to the queue, and when advertisements are lost we remove elements.
 // Each call to scan creates a cursor on the queue that will iterate until
 // the Listen is canceled.  We wait when we hit the end of the queue via the
 // condition variable cond.
 type inviteQueue struct {
 	mu   nsync.Mu
 	cond nsync.CV

 	elems        map[string]*ielement
 	sentinel     ielement
 	cursors      map[int]*ielement
 	nextCursorId int
 }

 func (q *inviteQueue) debugLocked() string {
 	buf := &bytes.Buffer{}
 	fmt.Fprintf(buf, "*%p", &q.sentinel)
 	for c := q.sentinel.next; c != &q.sentinel; c = c.next {
 		fmt.Fprintf(buf, " %p", c)
 	}
 	return buf.String()
 }

 type ielement struct {
 	invite     Invite
 	prev, next *ielement
 }

 func newInviteQueue() *inviteQueue {
 	iq := &inviteQueue{
 		elems:   make(map[string]*ielement),
 		cursors: make(map[int]*ielement),
 	}
 	iq.sentinel.next, iq.sentinel.prev = &iq.sentinel, &iq.sentinel
 	return iq
 }

 func (q *inviteQueue) empty() bool {
 	defer q.mu.Unlock()
 	q.mu.Lock()
 	return len(q.elems) == 0 && len(q.cursors) == 0
 }

 func (q *inviteQueue) size() int {
 	defer q.mu.Unlock()
 	q.mu.Lock()
 	return len(q.elems)
 }

 func (q *inviteQueue) remove(i Invite) {
 	defer q.mu.Unlock()
 	q.mu.Lock()

 	el, ok := q.elems[i.key]
 	if !ok {
 		return
 	}
 	delete(q.elems, i.key)
 	el.next.prev, el.prev.next = el.prev, el.next
 	for id, c := range q.cursors {
 		if c == el {
 			q.cursors[id] = c.prev
 		}
 	}
 }

 func (q *inviteQueue) add(i Invite) {
 	defer q.mu.Unlock()
 	q.mu.Lock()

 	if _, ok := q.elems[i.key]; ok {
 		return
 	}
 	el := &ielement{i, q.sentinel.prev, &q.sentinel}
 	q.sentinel.prev, q.sentinel.prev.next = el, el
 	q.elems[i.key] = el
 	q.cond.Broadcast()
 }

 func (q *inviteQueue) next(ctx *context.T, cursor int) (Invite, bool) {
 	defer q.mu.Unlock()
 	q.mu.Lock()
 	c, exists := q.cursors[cursor]
 	if !exists {
 		return Invite{}, false
 	}
 	for c.next == &q.sentinel {
 		if q.cond.WaitWithDeadline(&q.mu, nsync.NoDeadline, ctx.Done()) != nsync.OK {
 			delete(q.cursors, cursor)
 			return Invite{}, false
 		}
 		c = q.cursors[cursor]
 	}
 	c = c.next
 	q.cursors[cursor] = c
 	return c.invite, true
 }

 func (q *inviteQueue) scan() int {
 	defer q.mu.Unlock()
 	q.mu.Lock()
 	id := q.nextCursorId
 	q.nextCursorId++
 	q.cursors[id] = &q.sentinel
 	return id
 }

 func makeInvite(ad discovery.Advertisement) (Invite, wire.Id, bool) {
 	var dbId, sgId wire.Id
 	var ok bool
 	if dbId.Name, ok = ad.Attributes[wire.DiscoveryAttrDatabaseName]; !ok {
 		return Invite{}, dbId, false
 	}
 	if dbId.Blessing, ok = ad.Attributes[wire.DiscoveryAttrDatabaseBlessing]; !ok {
 		return Invite{}, dbId, false
 	}
 	if sgId.Name, ok = ad.Attributes[wire.DiscoveryAttrSyncgroupName]; !ok {
 		return Invite{}, dbId, false
 	}
 	if sgId.Blessing, ok = ad.Attributes[wire.DiscoveryAttrSyncgroupBlessing]; !ok {
 		return Invite{}, dbId, false
 	}
 	i := Invite{
 		Addresses: append([]string{}, ad.Addresses...),
 		Syncgroup: sgId,
 	}
 	sort.Strings(i.Addresses)
 	i.key = fmt.Sprintf("%v", i)
 	return i, dbId, true
 }

 func (i Invite) copy() Invite {
 	cp := i
 	cp.Addresses = append([]string(nil), i.Addresses...)
 	return cp
 }
	// Copyright 2016 The Vanadium Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style
	// license that can be found in the LICENSE file.

	package discovery

	import (
	"bytes"
	"fmt"
	"sort"
	"strings"
	"time"

	"v.io/v23"
	"v.io/v23/context"
	"v.io/v23/discovery"
	"v.io/v23/security"
	wire "v.io/v23/services/syncbase"
	"v.io/x/lib/nsync"
	"v.io/x/ref/lib/discovery/global"
	"v.io/x/ref/services/syncbase/server/interfaces"
	)

	const (
	visibilityKey = "vis"

	nhDiscoveryKey = iota
	globalDiscoveryKey
	)

	// Discovery implements v.io/v23/discovery.T for syncbase based
	// applications.
	// TODO(mattr): Actually this is not syncbase specific. At some
	// point we should just replace the result of v23.NewDiscovery
	// with this.
	type Discovery struct {
	nhDiscovery discovery.T
	globalDiscovery discovery.T
	}

	// NewDiscovery creates a new syncbase discovery object.
	// globalDiscoveryPath is the path in the namespace where global disovery
	// advertisements will be mounted.
	// If globalDiscoveryPath is empty, no global discovery service will be created.
	// globalScanInterval is the interval at which global discovery will be refreshed.
	// If globalScanInterval is 0, the defaultScanInterval of global discovery will
	// be used.
	func NewDiscovery(ctx *context.T, globalDiscoveryPath string, globalScanInterval time.Duration) (discovery.T, error) {
	d := &Discovery{}
	var err error
	if d.nhDiscovery, err = v23.NewDiscovery(ctx); err != nil {
	return nil, err
	}
	if globalDiscoveryPath != "" {
	if d.globalDiscovery, err = global.NewWithTTL(ctx, globalDiscoveryPath, 0, globalScanInterval); err != nil {
	return nil, err
	}
	}
	return d, nil
	}

	// Scan implements v.io/v23/discovery/T.Scan.
	func (d Discovery) Scan(ctx context.T, query string) (<-chan discovery.Update, error) {
	nhCtx, nhCancel := context.WithCancel(ctx)
	nhUpdates, err := d.nhDiscovery.Scan(nhCtx, query)
	if err != nil {
	nhCancel()
	return nil, err
	}
	var globalUpdates <-chan discovery.Update
	if d.globalDiscovery != nil {
	if globalUpdates, err = d.globalDiscovery.Scan(ctx, query); err != nil {
	nhCancel()
	return nil, err
	}
	}

	// Currently setting visibility on the neighborhood discovery
	// service turns IBE encryption on. We currently don't have the
	// infrastructure support for IBE, so that would make our advertisements
	// unreadable by everyone.
	// Instead we add the visibility list to the attributes of the advertisement
	// and filter on the client side. This is a temporary measure until
	// IBE is set up. See v.io/i/1345.
	updates := make(chan discovery.Update)
	go func() {
	defer close(updates)
	seen := make(map[discovery.AdId]*updateRef)
	for {
	var u discovery.Update
	var src uint // key of the source discovery service where the update came from
	select {
	case <-ctx.Done():
	return
	case u = <-nhUpdates:
	src = nhDiscoveryKey
	case u = <-globalUpdates:
	src = globalDiscoveryKey
	}
	d.handleUpdate(ctx, u, src, seen, updates)
	}
	}()

	return updates, nil
	}

	func (d Discovery) handleUpdate(ctx context.T, u discovery.Update, src uint, seen map[discovery.AdId]*updateRef, updates chan discovery.Update) {
	if u == nil {
	return
	}
	patterns := splitPatterns(u.Attribute(visibilityKey))
	if len(patterns) > 0 && !matchesPatterns(ctx, patterns) {
	return
	}

	id := u.Id()
	prev := seen[id]
	if u.IsLost() {
	// Only send the lost noitification if a found event was previously seen,
	// and all discovery services that found it have lost it.
	if prev == nil \|\| !prev.unset(src) {
	return
	}
	delete(seen, id)
	updates <- update{Update: u, lost: true}
	return
	}

	if prev == nil {
	// Always send updates for updates that we have never seen before.
	ref := &updateRef{update: u}
	ref.set(src)
	seen[id] = ref
	updates <- update{Update: u}
	return
	}

	if differ := updatesDiffer(prev.update, u); (differ && u.Timestamp().After(prev.update.Timestamp())) \|\|
	(!differ && src == nhDiscoveryKey && len(u.Advertisement().Attachments) > 0) {
	// If the updates differ and the newly found update has a later time than
	// previously found one, lose prev and find new.
	// Or, if the update doesn't differ, but is from neighborhood discovery, it
	// could have more information since we don't yet encode attachements in
	// global discovery.
	updates <- update{Update: prev.update, lost: true}
	ref := &updateRef{update: u}
	ref.set(src)
	seen[id] = ref
	updates <- update{Update: u}
	return
	}
	}

	// Advertise implements v.io/v23/discovery/T.Advertise.
	func (d Discovery) Advertise(ctx context.T, ad *discovery.Advertisement, visibility []security.BlessingPattern) (<-chan struct{}, error) {
	// Currently setting visibility on the neighborhood discovery
	// service turns IBE encryption on. We currently don't have the
	// infrastructure support for IBE, so that would make our advertisements
	// unreadable by everyone.
	// Instead we add the visibility list to the attributes of the advertisement
	// and filter on the client side. This is a temporary measure until
	// IBE is set up. See v.io/i/1345.
	adCopy := *ad
	if len(visibility) > 0 {
	adCopy.Attributes = make(discovery.Attributes, len(ad.Attributes)+1)
	for k, v := range ad.Attributes {
	adCopy.Attributes[k] = v
	}
	patterns := joinPatterns(visibility)
	adCopy.Attributes[visibilityKey] = patterns
	}

	stopped := make(chan struct{})
	nhCtx, nhCancel := context.WithCancel(ctx)
	nhStopped, err := d.nhDiscovery.Advertise(nhCtx, &adCopy, nil)
	if err != nil {
	nhCancel()
	return nil, err
	}
	var globalStopped <-chan struct{}
	if d.globalDiscovery != nil {
	if globalStopped, err = d.globalDiscovery.Advertise(ctx, &adCopy, nil); err != nil {
	nhCancel()
	<-nhStopped
	return nil, err
	}
	}
	go func() {
	<-nhStopped
	if d.globalDiscovery != nil {
	<-globalStopped
	}
	close(stopped)
	}()
	ad.Id = adCopy.Id
	return stopped, nil
	}

	func updatesDiffer(a, b discovery.Update) bool {
	if !sortedStringsEqual(a.Addresses(), b.Addresses()) {
	return true
	}
	if !mapsEqual(a.Advertisement().Attributes, b.Advertisement().Attributes) {
	return true
	}
	return false
	}

	func mapsEqual(a, b map[string]string) bool {
	if len(a) != len(b) {
	return false
	}
	for ka, va := range a {
	if vb, ok := b[ka]; !ok \|\| va != vb {
	return false
	}
	}
	return true
	}

	func sortedStringsEqual(a, b []string) bool {
	// We want to make a nil and an empty slices equal to avoid unnecessary inequality by that.
	if len(a) != len(b) {
	return false
	}
	for i, v := range a {
	if v != b[i] {
	return false
	}
	}
	return true
	}

	func matchesPatterns(ctx *context.T, patterns []security.BlessingPattern) bool {
	p := v23.GetPrincipal(ctx)
	blessings := p.BlessingStore().PeerBlessings()
	for _, b := range blessings {
	names := security.BlessingNames(p, b)
	for _, pattern := range patterns {
	if pattern.MatchedBy(names...) {
	return true
	}
	}
	}
	return false
	}

	type updateRef struct {
	update discovery.Update
	nhRef bool
	globalRef bool
	}

	func (r *updateRef) set(d uint) {
	switch d {
	case nhDiscoveryKey:
	r.nhRef = true
	case globalDiscoveryKey:
	r.globalRef = true
	}
	}

	func (r *updateRef) unset(d uint) bool {
	switch d {
	case nhDiscoveryKey:
	r.nhRef = false
	case globalDiscoveryKey:
	r.globalRef = false
	}
	return !r.nhRef && !r.globalRef
	}

	// update wraps the discovery.Update to remove the visibility attribute which we add
	// and allows us to mark the update as lost.
	type update struct {
	discovery.Update
	lost bool
	}

	func (u update) IsLost() bool { return u.lost }

	func (u update) Attribute(name string) string {
	if name == visibilityKey {
	return ""
	}
	return u.Update.Attribute(name)
	}

	func (u update) Advertisement() discovery.Advertisement {
	cp := u.Update.Advertisement()
	orig := cp.Attributes
	cp.Attributes = make(discovery.Attributes, len(orig))
	for k, v := range orig {
	if k != visibilityKey {
	cp.Attributes[k] = v
	}
	}
	return cp
	}

	// blessingSeparator is used to join multiple blessings into a
	// single string.
	// Note that comma cannot appear in blessings, see:
	// v.io/v23/security/certificate.go
	const blessingsSeparator = ','

	// joinPatterns concatenates the elements of a to create a single string.
	// The string can be split again with SplitPatterns.
	func joinPatterns(a []security.BlessingPattern) string {
	if len(a) == 0 {
	return ""
	}
	if len(a) == 1 {
	return string(a[0])
	}
	n := (len(a) - 1)
	for i := 0; i < len(a); i++ {
	n += len(a[i])
	}

	b := make([]byte, n)
	bp := copy(b, a[0])
	for _, s := range a[1:] {
	b[bp] = blessingsSeparator
	bp++
	bp += copy(b[bp:], s)
	}
	return string(b)
	}

	// splitPatterns splits BlessingPatterns that were joined with
	// JoinBlessingPattern.
	func splitPatterns(patterns string) []security.BlessingPattern {
	if patterns == "" {
	return nil
	}
	n := strings.Count(patterns, string(blessingsSeparator)) + 1
	out := make([]security.BlessingPattern, n)
	last, start := 0, 0
	for i, r := range patterns {
	if r == blessingsSeparator {
	out[last] = security.BlessingPattern(patterns[start:i])
	last++
	start = i + 1
	}
	}
	out[last] = security.BlessingPattern(patterns[start:])
	return out
	}

	var state struct {
	mu nsync.Mu
	scans map[security.Principal]*scanState
	}

	type scanState struct {
	dbs map[wire.Id]*inviteQueue
	listeners int
	cancel context.CancelFunc
	}

	func newScan(ctx context.T) (scanState, error) {
	ctx, cancel := context.WithRootCancel(ctx)
	scan := &scanState{
	dbs: make(map[wire.Id]*inviteQueue),
	cancel: cancel,
	}
	// TODO(suharshs): Add globalDiscoveryPath.
	d, err := NewDiscovery(ctx, "", 0)
	if err != nil {
	scan.cancel()
	return nil, err
	}
	query := fmt.Sprintf("v.InterfaceName=\"%s/%s\"",
	interfaces.SyncDesc.PkgPath, interfaces.SyncDesc.Name)
	updates, err := d.Scan(ctx, query)
	if err != nil {
	scan.cancel()
	return nil, err
	}
	go func() {
	for u := range updates {
	if invite, db, ok := makeInvite(u.Advertisement()); ok {
	state.mu.Lock()
	q := scan.dbs[db]
	if u.IsLost() {
	// TODO(mattr): Removing like this can result in resurfacing already
	// retrieved invites. For example if the ACL of a syncgroup is
	// changed to add a new memeber, then we might see a remove and then
	// later an add. In this case we would end up moving the invite
	// to the end of the queue. One way to fix this would be to
	// keep removed invites for some time.
	if q != nil {
	q.remove(invite)
	if q.empty() {
	delete(scan.dbs, db)
	}
	}
	} else {
	if q == nil {
	q = newInviteQueue()
	scan.dbs[db] = q
	}
	q.add(invite)
	q.cond.Broadcast()
	}
	state.mu.Unlock()
	}
	}
	}()
	return scan, nil
	}

	// Invite represents an invitation to join a syncgroup as found via Discovery.
	type Invite struct {
	Syncgroup wire.Id //Syncgroup is the Id of the syncgroup you've been invited to.
	Addresses []string //Addresses are the list of addresses of the inviting server.
	key string
	}

	// ListenForInvites listens via Discovery for syncgroup invitations for the given
	// database and sends the invites to the provided channel. We stop listening when
	// the given context is canceled. When that happens we close the given channel.
	func ListenForInvites(ctx *context.T, db wire.Id, ch chan<- Invite) error {
	defer state.mu.Unlock()
	state.mu.Lock()

	p := v23.GetPrincipal(ctx)
	scan := state.scans[p]
	if scan == nil {
	var err error
	if scan, err = newScan(ctx); err != nil {
	return err
	}
	if state.scans == nil {
	state.scans = make(map[security.Principal]*scanState)
	}
	state.scans[p] = scan
	}
	scan.listeners++

	q := scan.dbs[db]
	if q == nil {
	q = newInviteQueue()
	scan.dbs[db] = q
	}

	go func() {
	c := q.scan()
	for {
	next, ok := q.next(ctx, c)
	if !ok {
	break
	}
	select {
	case ch <- next.copy():
	case <-ctx.Done():
	break
	}
	}
	close(ch)

	defer state.mu.Unlock()
	state.mu.Lock()

	if q.empty() {
	delete(scan.dbs, db)
	}
	scan.listeners--
	if scan.listeners == 0 {
	scan.cancel()
	delete(state.scans, v23.GetPrincipal(ctx))
	}
	}()

	return nil
	}

	// inviteQueue is a linked list based queue. As we get new invitations we
	// add them to the queue, and when advertisements are lost we remove elements.
	// Each call to scan creates a cursor on the queue that will iterate until
	// the Listen is canceled. We wait when we hit the end of the queue via the
	// condition variable cond.
	type inviteQueue struct {
	mu nsync.Mu
	cond nsync.CV

	elems map[string]*ielement
	sentinel ielement
	cursors map[int]*ielement
	nextCursorId int
	}

	func (q *inviteQueue) debugLocked() string {
	buf := &bytes.Buffer{}
	fmt.Fprintf(buf, "*%p", &q.sentinel)
	for c := q.sentinel.next; c != &q.sentinel; c = c.next {
	fmt.Fprintf(buf, " %p", c)
	}
	return buf.String()
	}

	type ielement struct {
	invite Invite
	prev, next *ielement
	}

	func newInviteQueue() *inviteQueue {
	iq := &inviteQueue{
	elems: make(map[string]*ielement),
	cursors: make(map[int]*ielement),
	}
	iq.sentinel.next, iq.sentinel.prev = &iq.sentinel, &iq.sentinel
	return iq
	}

	func (q *inviteQueue) empty() bool {
	defer q.mu.Unlock()
	q.mu.Lock()
	return len(q.elems) == 0 && len(q.cursors) == 0
	}

	func (q *inviteQueue) size() int {
	defer q.mu.Unlock()
	q.mu.Lock()
	return len(q.elems)
	}

	func (q *inviteQueue) remove(i Invite) {
	defer q.mu.Unlock()
	q.mu.Lock()

	el, ok := q.elems[i.key]
	if !ok {
	return
	}
	delete(q.elems, i.key)
	el.next.prev, el.prev.next = el.prev, el.next
	for id, c := range q.cursors {
	if c == el {
	q.cursors[id] = c.prev
	}
	}
	}

	func (q *inviteQueue) add(i Invite) {
	defer q.mu.Unlock()
	q.mu.Lock()

	if _, ok := q.elems[i.key]; ok {
	return
	}
	el := &ielement{i, q.sentinel.prev, &q.sentinel}
	q.sentinel.prev, q.sentinel.prev.next = el, el
	q.elems[i.key] = el
	q.cond.Broadcast()
	}

	func (q inviteQueue) next(ctx context.T, cursor int) (Invite, bool) {
	defer q.mu.Unlock()
	q.mu.Lock()
	c, exists := q.cursors[cursor]
	if !exists {
	return Invite{}, false
	}
	for c.next == &q.sentinel {
	if q.cond.WaitWithDeadline(&q.mu, nsync.NoDeadline, ctx.Done()) != nsync.OK {
	delete(q.cursors, cursor)
	return Invite{}, false
	}
	c = q.cursors[cursor]
	}
	c = c.next
	q.cursors[cursor] = c
	return c.invite, true
	}

	func (q *inviteQueue) scan() int {
	defer q.mu.Unlock()
	q.mu.Lock()
	id := q.nextCursorId
	q.nextCursorId++
	q.cursors[id] = &q.sentinel
	return id
	}

	func makeInvite(ad discovery.Advertisement) (Invite, wire.Id, bool) {
	var dbId, sgId wire.Id
	var ok bool
	if dbId.Name, ok = ad.Attributes[wire.DiscoveryAttrDatabaseName]; !ok {
	return Invite{}, dbId, false
	}
	if dbId.Blessing, ok = ad.Attributes[wire.DiscoveryAttrDatabaseBlessing]; !ok {
	return Invite{}, dbId, false
	}
	if sgId.Name, ok = ad.Attributes[wire.DiscoveryAttrSyncgroupName]; !ok {
	return Invite{}, dbId, false
	}
	if sgId.Blessing, ok = ad.Attributes[wire.DiscoveryAttrSyncgroupBlessing]; !ok {
	return Invite{}, dbId, false
	}
	i := Invite{
	Addresses: append([]string{}, ad.Addresses...),
	Syncgroup: sgId,
	}
	sort.Strings(i.Addresses)
	i.key = fmt.Sprintf("%v", i)
	return i, dbId, true
	}

	func (i Invite) copy() Invite {
	cp := i
	cp.Addresses = append([]string(nil), i.Addresses...)
	return cp
	}