Files
zeekay 9f22e43b50 rpc: real Target-based promise pipelining (Session + Pipeliner)
The call envelope has always carried Target @8, but zapgen hardcoded
Target: rpc.NoTarget in every generated method, so Go could DECODE a
pipelined envelope yet never ORIGINATE or RESOLVE one — pipelining was a
no-op. Make the canonical Target model real on both sides.

Client (origination): rpc.Session allocates unique non-zero PromiseIDs and
builds Calls — Origin (Target=NoTarget) and Pipeline (Target=prior.ID).
zapgen now emits a *rpc.Session-backed client: M(...) originates, MOn(on
rpc.Promise) pipelines on a prior answer (payload supplied server-side).
NoTarget is no longer hardcoded — it flows through one shared invoke<M>
path. Both forms return the call's own rpc.Promise so chains compose.

Server (resolution): rpc.Pipeliner wraps a generated Dispatch<Iface> and is
the per-connection promise table: records each OK answer under its
PromiseID, substitutes a resolved Target's Body for the dependent's Payload
before dispatch, QUEUES a dependent whose Target hasn't resolved yet until
it does, and REFUSES (StatusBadRequest, never hangs) a dependent whose
Target answered non-OK or was Finished. Finish(id) bounds the table.

Wire-compatible: Target=NoTarget is the non-pipelined call, so a Pipeliner
and a plain dispatcher are byte-identical on the wire; a non-pipelining
peer (any other runtime) interoperates trivially.

Proven end-to-end in rpc/pipeline_test.go (auth->getResource): single-pass
resolution, server-side queuing when the dependent arrives first, a 3-deep
A->B->C chain, non-OK/Finished targets refused, Finish waking a parked
dependent, and Target riding on the wire byte-for-byte. Full suite green
under -race. Docs: README + LLM.md describe the one canonical model and the
Rust capnp PromisedAnswer boundary honestly.
2026-06-21 01:49:48 -07:00

199 lines
6.0 KiB
Go

// Code generated by zapgen; DO NOT EDIT.
// source: echo.zap
package echo
import (
"fmt"
"github.com/zap-proto/go/rpc"
)
// Method ordinals for the Echo service (stable 1-based wire ids).
const (
EchoPingOrdinal uint32 = 1
EchoNotifyOrdinal uint32 = 2
EchoHealthOrdinal uint32 = 3
EchoShutdownOrdinal uint32 = 4
)
// EchoChannel ships one Call envelope and awaits its correlated Response.
type EchoChannel interface {
Call(envelope []byte) (rpc.Response, error)
}
// EchoClient is a typed RPC client for the Echo service over a ZAP call channel.
// Each call takes a fresh PromiseID from sess; the pipelined "On" form of a
// method sets Target to a prior call's Promise so the server chains them.
type EchoClient struct {
ch EchoChannel
cap []byte
sess *rpc.Session
}
// NewEchoClient returns a client that issues calls over ch, attaching cap
// (which may be nil) to every request.
func NewEchoClient(ch EchoChannel, capability []byte) *EchoClient {
return &EchoClient{ch: ch, cap: capability, sess: rpc.NewSession()}
}
func (c *EchoClient) Ping(req []byte) (rpc.Promise, []byte, error) {
return c.invokePing(rpc.NoTarget, req)
}
// PingOn issues Ping as a dependent call pipelined on the answer of on:
// the server substitutes on's resolved result for this call's payload
// before dispatch, so it ships without waiting for on to round-trip.
func (c *EchoClient) PingOn(on rpc.Promise) (rpc.Promise, []byte, error) {
return c.invokePing(on.ID, nil)
}
func (c *EchoClient) invokePing(target uint32, payload []byte) (rpc.Promise, []byte, error) {
p := c.sess.Next()
resp, err := c.ch.Call(rpc.BuildRequest(rpc.Call{
Method: EchoPingOrdinal,
PromiseID: p.ID,
Target: target,
Cap: c.cap,
Payload: payload,
}))
if err != nil {
return p, nil, err
}
if resp.Status != rpc.StatusOK {
return p, nil, fmt.Errorf("Echo.Ping: status %d", resp.Status)
}
return p, resp.Body, nil
}
func (c *EchoClient) Notify(req []byte) (rpc.Promise, error) {
return c.invokeNotify(rpc.NoTarget, req)
}
// NotifyOn issues Notify as a dependent call pipelined on the answer of on:
// the server substitutes on's resolved result for this call's payload
// before dispatch, so it ships without waiting for on to round-trip.
func (c *EchoClient) NotifyOn(on rpc.Promise) (rpc.Promise, error) {
return c.invokeNotify(on.ID, nil)
}
func (c *EchoClient) invokeNotify(target uint32, payload []byte) (rpc.Promise, error) {
p := c.sess.Next()
resp, err := c.ch.Call(rpc.BuildRequest(rpc.Call{
Method: EchoNotifyOrdinal,
PromiseID: p.ID,
Target: target,
Cap: c.cap,
Payload: payload,
}))
if err != nil {
return p, err
}
if resp.Status != rpc.StatusOK {
return p, fmt.Errorf("Echo.Notify: status %d", resp.Status)
}
return p, nil
}
func (c *EchoClient) Health() (rpc.Promise, []byte, error) {
return c.invokeHealth(rpc.NoTarget, nil)
}
// HealthOn issues Health as a dependent call pipelined on the answer of on:
// the server substitutes on's resolved result for this call's payload
// before dispatch, so it ships without waiting for on to round-trip.
func (c *EchoClient) HealthOn(on rpc.Promise) (rpc.Promise, []byte, error) {
return c.invokeHealth(on.ID, nil)
}
func (c *EchoClient) invokeHealth(target uint32, payload []byte) (rpc.Promise, []byte, error) {
p := c.sess.Next()
resp, err := c.ch.Call(rpc.BuildRequest(rpc.Call{
Method: EchoHealthOrdinal,
PromiseID: p.ID,
Target: target,
Cap: c.cap,
Payload: payload,
}))
if err != nil {
return p, nil, err
}
if resp.Status != rpc.StatusOK {
return p, nil, fmt.Errorf("Echo.Health: status %d", resp.Status)
}
return p, resp.Body, nil
}
func (c *EchoClient) Shutdown() (rpc.Promise, error) {
return c.invokeShutdown(rpc.NoTarget, nil)
}
// ShutdownOn issues Shutdown as a dependent call pipelined on the answer of on:
// the server substitutes on's resolved result for this call's payload
// before dispatch, so it ships without waiting for on to round-trip.
func (c *EchoClient) ShutdownOn(on rpc.Promise) (rpc.Promise, error) {
return c.invokeShutdown(on.ID, nil)
}
func (c *EchoClient) invokeShutdown(target uint32, payload []byte) (rpc.Promise, error) {
p := c.sess.Next()
resp, err := c.ch.Call(rpc.BuildRequest(rpc.Call{
Method: EchoShutdownOrdinal,
PromiseID: p.ID,
Target: target,
Cap: c.cap,
Payload: payload,
}))
if err != nil {
return p, err
}
if resp.Status != rpc.StatusOK {
return p, fmt.Errorf("Echo.Shutdown: status %d", resp.Status)
}
return p, nil
}
// EchoHandler is the server contract for the Echo service. Implement each
// method, then route requests to it with DispatchEcho.
type EchoHandler interface {
Ping(req []byte) ([]byte, error)
Notify(req []byte) error
Health() ([]byte, error)
Shutdown() error
}
// DispatchEcho decodes a Call envelope, routes it by method ordinal to h,
// and returns the response envelope. An unknown ordinal yields a
// StatusNotFound response; a handler error yields StatusInternal.
func DispatchEcho(h EchoHandler, envelope []byte) ([]byte, error) {
call, err := rpc.ParseRequest(envelope)
if err != nil {
return nil, err
}
switch call.Method {
case EchoPingOrdinal:
body, err := h.Ping(call.Payload)
if err != nil {
return rpc.BuildResponse(rpc.StatusInternal, call.PromiseID, nil), nil
}
return rpc.BuildResponse(rpc.StatusOK, call.PromiseID, body), nil
case EchoNotifyOrdinal:
if err := h.Notify(call.Payload); err != nil {
return rpc.BuildResponse(rpc.StatusInternal, call.PromiseID, nil), nil
}
return rpc.BuildResponse(rpc.StatusOK, call.PromiseID, nil), nil
case EchoHealthOrdinal:
body, err := h.Health()
if err != nil {
return rpc.BuildResponse(rpc.StatusInternal, call.PromiseID, nil), nil
}
return rpc.BuildResponse(rpc.StatusOK, call.PromiseID, body), nil
case EchoShutdownOrdinal:
if err := h.Shutdown(); err != nil {
return rpc.BuildResponse(rpc.StatusInternal, call.PromiseID, nil), nil
}
return rpc.BuildResponse(rpc.StatusOK, call.PromiseID, nil), nil
default:
return rpc.BuildResponse(rpc.StatusNotFound, call.PromiseID, nil), nil
}
}