Handle incoming resolve messages.

localpromise.go

@@ -5,15 +5,15 @@ import (
 )
 
 // ClientHook for a promise that will be resolved to some other capability
-// at some point. Buffers calls in a queue until the promsie is fulfilled,
+// at some point. Buffers calls in a queue until the promise is fulfilled,
 // then forwards them.
 type localPromise struct {
 	aq *AnswerQueue
 }
 
 // NewLocalPromise returns a client that will eventually resolve to a capability,
-// supplied via the fulfiller.
-func NewLocalPromise[C ~ClientKind]() (C, Resolver[C]) {
+// supplied via resolver. resolver.Fulfill steals the reference to its argument.
+func NewLocalPromise[C ~ClientKind]() (promise C, resolver Resolver[C]) {
 	lp := newLocalPromise()
 	p, f := NewPromisedClient(lp)
 	return C(p), localResolver[C]{
@@ -44,27 +44,38 @@ func (lp localPromise) String() string {
 	return "localPromise{...}"
 }
 
-func (lp localPromise) Fulfill(c Client) {
-	msg, seg := NewSingleSegmentMessage(nil)
-	capID := msg.AddCap(c)
-	lp.aq.Fulfill(NewInterface(seg, capID).ToPtr())
-}
-
-func (lp localPromise) Reject(err error) {
-	lp.aq.Reject(err)
-}
-
 type localResolver[C ~ClientKind] struct {
 	lp             localPromise
 	clientResolver Resolver[Client]
 }
 
 func (lf localResolver[C]) Fulfill(c C) {
-	lf.lp.Fulfill(Client(c))
-	lf.clientResolver.Fulfill(Client(c))
+	// This is convoluted, for a few reasons:
+	//
+	// 1. AnswerQueue wants a Ptr, not a Client, so we have to construct
+	//    a message for this.
+	// 2. Message.AddCap steals the reference, so we have to get the client
+	//    back from the message instead of using the reference we already
+	//    have.
+	// 3. The semantics of NewPromisedClient differ from what we want, and
+	//    are kindof odd: when it is resolved it does not steal the reference,
+	//    nor does it borrow it -- instead it merges the refcounts, so that
+	//    the two clients point to the same place. So we have to drop our
+	//    reference to get the semantics we want.
+	//
+	//    TODO: We should probably push this part down into the implementation of
+	//    clientPromise. That requires auditing its uses and adjusting call sites
+	//    though.
+	msg, seg := NewSingleSegmentMessage(nil)
+	capID := msg.AddCap(Client(c))
+	iface := NewInterface(seg, capID)
+	client := iface.Client()
+	defer client.Release()
+	lf.lp.aq.Fulfill(iface.ToPtr())
+	lf.clientResolver.Fulfill(client)
 }
 
 func (lf localResolver[C]) Reject(err error) {
-	lf.lp.Reject(err)
+	lf.lp.aq.Reject(err)
 	lf.clientResolver.Reject(err)
 }

rpc/export.go

@@ -74,9 +74,11 @@ func (c *lockedConn) releaseExport(id exportID, count uint32) (capnp.Client, err
 		c.lk.exports[id] = nil
 		c.lk.exportID.remove(uint32(id))
 		metadata := client.State().Metadata
-		syncutil.With(metadata, func() {
-			c.clearExportID(metadata)
-		})
+		if metadata != nil {
+			syncutil.With(metadata, func() {
+				c.clearExportID(metadata)
+			})
+		}
 		return client, nil
 	case count > ent.wireRefs:
 		return capnp.Client{}, rpcerr.Failed(errors.New("export ID " + str.Utod(id) + " released too many references"))
@@ -367,33 +369,39 @@ func (e *embargo) Shutdown() {
 // senderLoopback holds the salient information for a sender-loopback
 // Disembargo message.
 type senderLoopback struct {
-	id        embargoID
-	question  questionID
-	transform []capnp.PipelineOp
+	id     embargoID
+	target parsedMessageTarget
 }
 
 func (sl *senderLoopback) buildDisembargo(msg rpccp.Message) error {
 	d, err := msg.NewDisembargo()
 	if err != nil {
 		return rpcerr.WrapFailed("build disembargo", err)
 	}
+	d.Context().SetSenderLoopback(uint32(sl.id))
 	tgt, err := d.NewTarget()
 	if err != nil {
 		return rpcerr.WrapFailed("build disembargo", err)
 	}
-	pa, err := tgt.NewPromisedAnswer()
-	if err != nil {
-		return rpcerr.WrapFailed("build disembargo", err)
-	}
-	oplist, err := pa.NewTransform(int32(len(sl.transform)))
-	if err != nil {
-		return rpcerr.WrapFailed("build disembargo", err)
-	}
+	switch sl.target.which {
+	case rpccp.MessageTarget_Which_promisedAnswer:
+		pa, err := tgt.NewPromisedAnswer()
+		if err != nil {
+			return rpcerr.WrapFailed("build disembargo", err)
+		}
+		oplist, err := pa.NewTransform(int32(len(sl.target.transform)))
+		if err != nil {
+			return rpcerr.WrapFailed("build disembargo", err)
+		}
 
-	d.Context().SetSenderLoopback(uint32(sl.id))
-	pa.SetQuestionId(uint32(sl.question))
-	for i, op := range sl.transform {
-		oplist.At(i).SetGetPointerField(op.Field)
+		pa.SetQuestionId(uint32(sl.target.promisedAnswer))
+		for i, op := range sl.target.transform {
+			oplist.At(i).SetGetPointerField(op.Field)
+		}
+	case rpccp.MessageTarget_Which_importedCap:
+		tgt.SetImportedCap(uint32(sl.target.importedCap))
+	default:
+		return errors.New("unknown variant for MessageTarget: " + str.Utod(sl.target.which))
 	}
 	return nil
 }

rpc/import.go

@@ -45,14 +45,19 @@ type impent struct {
 	// importClient's generation matches the entry's generation before
 	// removing the entry from the table and sending a release message.
 	generation uint64
+
+	// If resolver is non-nil, then this is a promise (received as
+	// CapDescriptor_Which_senderPromise), and when a resolve message
+	// arrives we should use this to fulfill the promise locally.
+	resolver capnp.Resolver[capnp.Client]
 }
 
 // addImport returns a client that represents the given import,
 // incrementing the number of references to this import from this vat.
 // This is separate from the reference counting that capnp.Client does.
 //
 // The caller must be holding onto c.mu.
-func (c *lockedConn) addImport(id importID) capnp.Client {
+func (c *lockedConn) addImport(id importID, isPromise bool) capnp.Client {
 	if ent := c.lk.imports[id]; ent != nil {
 		ent.wireRefs++
 		client, ok := ent.wc.AddRef()
@@ -67,13 +72,23 @@ func (c *lockedConn) addImport(id importID) capnp.Client {
 		}
 		return client
 	}
-	client := capnp.NewClient(&importClient{
+	hook := &importClient{
 		c:  (*Conn)(c),
 		id: id,
-	})
+	}
+	var (
+		client   capnp.Client
+		resolver capnp.Resolver[capnp.Client]
+	)
+	if isPromise {
+		client, resolver = capnp.NewPromisedClient(hook)
+	} else {
+		client = capnp.NewClient(hook)
+	}
 	c.lk.imports[id] = &impent{
 		wc:       client.WeakRef(),
 		wireRefs: 1,
+		resolver: resolver,
 	}
 	return client
 }

rpc/localpromise_test.go

@@ -45,12 +45,10 @@ func TestLocalPromiseFulfill(t *testing.T) {
 	})
 	defer rel2()
 
-	pp := testcapnp.PingPong_ServerToClient(&echoNumOrderChecker{
+	r.Fulfill(testcapnp.PingPong_ServerToClient(&echoNumOrderChecker{
 		t:       t,
 		nextNum: 1,
-	})
-	defer pp.Release()
-	r.Fulfill(pp)
+	}))
 
 	fut3, rel3 := p.EchoNum(ctx, func(p testcapnp.PingPong_echoNum_Params) error {
 		p.SetN(3)
@@ -70,31 +68,29 @@ func TestLocalPromiseFulfill(t *testing.T) {
 	assert.Equal(t, int64(3), res3.N())
 }
 
+func echoNum(ctx context.Context, pp testcapnp.PingPong, n int64) (testcapnp.PingPong_echoNum_Results_Future, capnp.ReleaseFunc) {
+	return pp.EchoNum(ctx, func(p testcapnp.PingPong_echoNum_Params) error {
+		p.SetN(n)
+		return nil
+	})
+}
+
 func TestLocalPromiseReject(t *testing.T) {
 	t.Parallel()
 
 	ctx := context.Background()
 	p, r := capnp.NewLocalPromise[testcapnp.PingPong]()
 	defer p.Release()
 
-	fut1, rel1 := p.EchoNum(ctx, func(p testcapnp.PingPong_echoNum_Params) error {
-		p.SetN(1)
-		return nil
-	})
+	fut1, rel1 := echoNum(ctx, p, 1)
 	defer rel1()
 
-	fut2, rel2 := p.EchoNum(ctx, func(p testcapnp.PingPong_echoNum_Params) error {
-		p.SetN(2)
-		return nil
-	})
+	fut2, rel2 := echoNum(ctx, p, 2)
 	defer rel2()
 
 	r.Reject(errors.New("Promise rejected"))
 
-	fut3, rel3 := p.EchoNum(ctx, func(p testcapnp.PingPong_echoNum_Params) error {
-		p.SetN(3)
-		return nil
-	})
+	fut3, rel3 := echoNum(ctx, p, 3)
 	defer rel3()
 
 	_, err := fut1.Struct()
@@ -106,3 +102,17 @@ func TestLocalPromiseReject(t *testing.T) {
 	_, err = fut3.Struct()
 	assert.NotNil(t, err)
 }
+
+// Test that the promise owns the capability it resolves to; no separate
+// release should be necessary.
+func TestLocalPromiseOwnsResult(t *testing.T) {
+	t.Parallel()
+
+	p, r := capnp.NewLocalPromise[testcapnp.PingPong]()
+	defer p.Release()
+
+	r.Fulfill(testcapnp.PingPong_ServerToClient(&echoNumOrderChecker{
+		t:       t,
+		nextNum: 1,
+	}))
+}