Handle incoming resolve messages.

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

@@ -70,31 +70,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()