code: func (ce *cacheEntry) registerStream(ctx context.Context, coord Coordinator, st LogStreamState, desc []byte) (*LogStreamState, error) {
  	// Initialize the registration Promise, if one is not defined.
  	//
  	// While locked, load the current registration promise and the local
  	// terminal index value.
  	var (
  		p    *promise.Promise
  		tidx types.MessageIndex
  	)
  	ce.withLock(func() {
  		if ce.registerP == nil {
  			ce.registerP = promise.NewDeferred(func(ctx context.Context) (interface{}, error) {
  				st, err := coord.RegisterStream(ctx, &st, desc)
  				if err == nil {
  					// If the remote state has a terminal index, retain it locally.
  					ce.loadRemoteTerminalIndex(st.TerminalIndex)
  					return st, nil
  				}
  
  				return nil, err
  			})
  		}
  
  		p, tidx = ce.registerP, ce.terminalIndex
  	})
  
  	// Resolve our registration Promise.
  	remoteStateIface, err := p.Get(ctx)
  	if err != nil {
  		// If the promise failed transiently, clear it so that subsequent callers
  		// will regenerate a new promise. ONLY clear it if it it is the same
  		// promise, as different callers may have already cleared/rengerated it.
  		if transient.Tag.In(err) {
  			ce.withLock(func() {
  				if ce.registerP == p {
  					ce.registerP = nil
  				}
  			})
  		}
  		return nil, err
  	}
  	remoteState := remoteStateIface.(*LogStreamState)
  
  	// If our remote state doesn't include a terminal index and our local state
  	// has recorded a successful remote terminal index, return a copy of the
  	// remote state with the remote terminal index added.
  	if remoteState.TerminalIndex < 0 && tidx >= 0 {
  		remoteStateCopy := *remoteState
  		remoteStateCopy.TerminalIndex = tidx
  		remoteState = &remoteStateCopy
  	}
  	return remoteState, nil
  }
docs: // registerStream performs a RegisterStream Coordinator RPC.