The compose ask carries its request gzip-packed; requests of another version are refused by name (issue 1966)
A large node's request is 1.2 MB, over the bus's 1 MiB message limit, so the compose ask failed once keeping no longer did. The checker packs the request (a JSON string "gzip+base64:..." in the part's `any` request, no contract change) and buildstep.ComposeJob.RequestBytes reads either form. A kept request of another version is refused naming it, and listing no longer reads pieces. An older mesh-build-go refuses a packed ask loudly (it holds no request object): the check fails naming it, so the catalog change is delivered with this one as a group. Quick fix (chosen as the recovery): pack the request, ~290 KB for the largest node. Thorough fix (milestone G): the request named by digest and served on a seat verb, as RunInput is. Claude-Session: https://claude.ai/code/session_01NcbgfFfQ4Z9UE5jK32PmE5
This commit is contained in:
@@ -61,6 +61,21 @@ func TestTheToolsServedAreTheFiveReadTools(t *testing.T) {
|
||||
if one, err := rt[0].Run(map[string]any{"check": "big.0123456789ab", "node": "anchor"}); err != nil || !strings.Contains(func() string { b, _ := json.Marshal(one); return string(b) }(), base64.StdEncoding.EncodeToString(rnd)[:64]) {
|
||||
t.Errorf("the big request read back: %v", err)
|
||||
}
|
||||
// Listing reads no piece, and a request of another version is refused by name.
|
||||
store.vals["v1.0123456789ab/anchor"] = json.RawMessage(`{"v":1,"check":"v1.0123456789ab","node":"anchor","gzip":"x"}`)
|
||||
noPieces := check.StateRead{Keys: store.Keys, Get: func(k string) (json.RawMessage, error) {
|
||||
if strings.HasPrefix(k, "pieces/") {
|
||||
return nil, errors.New("a piece was read")
|
||||
}
|
||||
return store.Get(k)
|
||||
}}
|
||||
if _, err := RequestTools(noPieces)[0].Run(map[string]any{}); err != nil {
|
||||
t.Errorf("listing: %v", err)
|
||||
}
|
||||
if _, err := rt[0].Run(map[string]any{"check": "v1.0123456789ab", "node": "anchor"}); err == nil || !strings.Contains(err.Error(), "version 1") {
|
||||
t.Errorf("a version 1 request: %v", err)
|
||||
}
|
||||
delete(store.vals, "v1.0123456789ab/anchor")
|
||||
if all, _ := rt[0].Run(map[string]any{}); strings.Count(func() string { b, _ := json.Marshal(all); return string(b) }(), `"node"`) != 4 {
|
||||
t.Errorf("the list holds a piece as a request: %v", all)
|
||||
}
|
||||
|
||||
@@ -228,6 +228,10 @@ func (e EngineCompositions) inSandbox(ctx context.Context, c *Case, head, node s
|
||||
}
|
||||
job := buildstep.ComposeJob{Fetch: buildstep.Fetch{Repository: nodeEngineRepo, Commit: head, Packages: packagesFor(e.Published, c, c.Tagged, nodeEngineRepo)}, Dir: meshcatalogclient.RootPath,
|
||||
Package: composer.Package, Request: raw, Reader: "the node-engine head of " + c.Group}
|
||||
// The ask travels on the bus, whose message limit a whole request passes (1.2 MB for a large node): it goes packed.
|
||||
if err := job.PackRequest(raw); err != nil {
|
||||
return a, err
|
||||
}
|
||||
verdict, err := e.Sandbox.Compose(ctx, job)
|
||||
if err != nil {
|
||||
return a, fmt.Errorf("the composer at the node-engine's head %s did not answer for %s: %w", head, node, err)
|
||||
|
||||
@@ -59,7 +59,11 @@ type fakeSandbox struct {
|
||||
|
||||
func (f *fakeSandbox) Compose(_ context.Context, job buildstep.ComposeJob) (composer.Verdict, error) {
|
||||
var req composer.Request
|
||||
if err := json.Unmarshal(job.Request, &req); err != nil {
|
||||
body, err := job.RequestBytes()
|
||||
if err != nil {
|
||||
return composer.Verdict{}, err
|
||||
}
|
||||
if err := json.Unmarshal(body, &req); err != nil {
|
||||
return composer.Verdict{}, err
|
||||
}
|
||||
f.mu.Lock()
|
||||
@@ -206,7 +210,11 @@ func TestTheCompositionIsMadeFromTheEnginesAnswers(t *testing.T) {
|
||||
}
|
||||
job := sb.jobs[0]
|
||||
var req composer.Request
|
||||
_ = json.Unmarshal(job.Request, &req)
|
||||
if len(job.Request) == 0 || job.Request[0] != '"' {
|
||||
t.Errorf("the request went whole, not packed (issue 1966): %.60s", job.Request)
|
||||
}
|
||||
body, _ := job.RequestBytes()
|
||||
_ = json.Unmarshal(body, &req)
|
||||
if job.Fetch.Repository != nodeEngineRepo || job.Fetch.Commit != head2 || job.Package != composer.Package ||
|
||||
string(req.Input) != `{"nodes":[{"name":"anchor"}]}` || req.Cases[0].Manifests["web"].Digest == "" ||
|
||||
len(req.Seats) != 1 || string(req.Seats[0]) != string(host.Seats[0]) {
|
||||
@@ -220,8 +228,9 @@ func TestTheCompositionIsMadeFromTheEnginesAnswers(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
keyOfHost, _ := nodeengineclient.CompositionKey(host.Group, host.Heads)
|
||||
lastBody, _ := sb.jobs[len(sb.jobs)-1].RequestBytes()
|
||||
got, err := ReadRequests(store.read(), keyOfHost, "anchor")
|
||||
if b, _ := json.Marshal(got); err != nil || !strings.Contains(string(b), `"request":`+string(sb.jobs[len(sb.jobs)-1].Request)) {
|
||||
if b, _ := json.Marshal(got); err != nil || !strings.Contains(string(b), `"request":`+string(lastBody)) {
|
||||
t.Errorf("kept %s, %v: want the request of the last job under %s/anchor", b, err, keyOfHost)
|
||||
}
|
||||
store.err = errors.New("the state is full")
|
||||
@@ -311,7 +320,8 @@ func TestTheCompositionIsMadeFromTheEnginesAnswers(t *testing.T) {
|
||||
t.Fatalf("jobs %+v: want one job at the node-engine's trunk", sb.jobs)
|
||||
}
|
||||
var req composer.Request
|
||||
_ = json.Unmarshal(sb.jobs[0].Request, &req)
|
||||
b0, _ := sb.jobs[0].RequestBytes()
|
||||
_ = json.Unmarshal(b0, &req)
|
||||
if got := string(req.Cases[0].Verdicts["mesh-build-go"]); got != string(brings.Judged["mesh-build-go"]) || len(req.Cases[0].Verdicts) != 2 {
|
||||
t.Errorf("the request's verdicts %s: want the check's judge's for mesh-build-go and the indexer's for mail", req.Cases[0].Verdicts)
|
||||
}
|
||||
@@ -321,7 +331,8 @@ func TestTheCompositionIsMadeFromTheEnginesAnswers(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var none composer.Request
|
||||
_ = json.Unmarshal(sb.jobs[len(sb.jobs)-1].Request, &none)
|
||||
bn, _ := sb.jobs[len(sb.jobs)-1].RequestBytes()
|
||||
_ = json.Unmarshal(bn, &none)
|
||||
if _, kept := none.Cases[0].Verdicts["mesh-build-go"]; kept {
|
||||
t.Errorf("the running indexer's verdict on the change's manifest was kept: %s", none.Cases[0].Verdicts)
|
||||
}
|
||||
|
||||
@@ -62,6 +62,16 @@ type KeptPiece struct {
|
||||
Gzip string `json:"gzip"`
|
||||
}
|
||||
|
||||
// isPieceKey is whether a key is a piece's: under pieces/ and a sha256 digest, which no check's key (group.heads/node) is.
|
||||
func isPieceKey(k string) bool {
|
||||
rest, ok := strings.CutPrefix(k, "pieces/")
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
digest, _, _ := strings.Cut(rest, "/")
|
||||
return len(digest) == 64
|
||||
}
|
||||
|
||||
// RequestChunkKey is the key of a request's piece: under the request's digest, so a request kept again for the same
|
||||
// bytes writes the same pieces.
|
||||
func RequestChunkKey(digest string, index int) string {
|
||||
@@ -156,6 +166,9 @@ func ReadRequests(s StateRead, key, node string) (any, error) {
|
||||
if key != "" && !strings.HasPrefix(k, key+"/") {
|
||||
continue
|
||||
}
|
||||
if isPieceKey(k) {
|
||||
continue // a piece of a request, not a request: never read here
|
||||
}
|
||||
raw, err := s.Get(k)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -165,9 +178,6 @@ func ReadRequests(s StateRead, key, node string) (any, error) {
|
||||
listed = append(listed, map[string]any{"key": k, "unreadable": err.Error()})
|
||||
continue
|
||||
}
|
||||
if kept.Check == "" {
|
||||
continue // a piece of a request, not a request
|
||||
}
|
||||
listed = append(listed, map[string]any{"check": kept.Check, "node": kept.Node, "size": kept.Size, "digest": kept.Digest, "kept_at": kept.KeptAt})
|
||||
}
|
||||
return map[string]any{"kept": listed}, nil
|
||||
@@ -179,6 +189,9 @@ func unpackRequest(s StateRead, raw []byte) (KeptRequest, []byte, error) {
|
||||
if err := json.Unmarshal(raw, &kept); err != nil {
|
||||
return kept, nil, fmt.Errorf("is no kept request: %w", err)
|
||||
}
|
||||
if kept.V != 2 {
|
||||
return kept, nil, fmt.Errorf("is a kept request of version %d, which this reader does not read (it reads version 2; an older one is past its lifetime within 3 days)", kept.V)
|
||||
}
|
||||
var packed []byte
|
||||
for i := 0; i < kept.Pieces; i++ {
|
||||
raw, err := s.Get(RequestChunkKey(kept.Digest, i))
|
||||
|
||||
+53
-4
@@ -1,8 +1,12 @@
|
||||
package buildstep
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"compress/gzip"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"path"
|
||||
"slices"
|
||||
"sort"
|
||||
@@ -281,13 +285,54 @@ type CheckJob struct {
|
||||
|
||||
// ComposeJob asks one compose-candidate build.
|
||||
type ComposeJob struct {
|
||||
Fetch Fetch `json:"fetch"`
|
||||
Dir string `json:"dir"`
|
||||
Package string `json:"package"`
|
||||
Fetch Fetch `json:"fetch"`
|
||||
Dir string `json:"dir"`
|
||||
Package string `json:"package"`
|
||||
// Request is the composer's request: a JSON object, or, for one near the bus's message limit (issue 1966), a JSON
|
||||
// string PackedRequestPrefix + the base64 of the object gzip-compressed. The part types it `any`, so both fit it; a
|
||||
// build module that does not know the packed form refuses the ask naming that it holds no request object.
|
||||
Request json.RawMessage `json:"request"`
|
||||
Reader string `json:"reader,omitempty"`
|
||||
}
|
||||
|
||||
// PackedRequestPrefix opens a compose job's request that travels gzip-compressed.
|
||||
const PackedRequestPrefix = "gzip+base64:"
|
||||
|
||||
// PackRequest sets the job's request from the JSON object raw, packed.
|
||||
func (c *ComposeJob) PackRequest(raw []byte) error {
|
||||
var b bytes.Buffer
|
||||
zw := gzip.NewWriter(&b)
|
||||
if _, err := zw.Write(raw); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := zw.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
packed, err := json.Marshal(PackedRequestPrefix + base64.StdEncoding.EncodeToString(b.Bytes()))
|
||||
c.Request = packed
|
||||
return err
|
||||
}
|
||||
|
||||
// RequestBytes is the job's request as a JSON object, unpacked when it travels packed.
|
||||
func (c ComposeJob) RequestBytes() ([]byte, error) {
|
||||
var packed string
|
||||
if len(c.Request) == 0 || c.Request[0] != '"' {
|
||||
return c.Request, nil
|
||||
}
|
||||
if err := json.Unmarshal(c.Request, &packed); err != nil || !strings.HasPrefix(packed, PackedRequestPrefix) {
|
||||
return nil, fmt.Errorf("buildstep: a compose job's request is a string that is no packed request")
|
||||
}
|
||||
raw, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(packed, PackedRequestPrefix))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("buildstep: the packed request of a compose job is no base64: %w", err)
|
||||
}
|
||||
zr, err := gzip.NewReader(bytes.NewReader(raw))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("buildstep: the packed request of a compose job is no gzip: %w", err)
|
||||
}
|
||||
return io.ReadAll(zr)
|
||||
}
|
||||
|
||||
// AskOf is any seat's ask (or this form) as the neutral StepAsk: the one conversion, since the three seats' types
|
||||
// are the part's one layout. A value that does not fit the layout is refused.
|
||||
func AskOf[T any](seatAsk T) (StepAsk, error) {
|
||||
@@ -378,7 +423,11 @@ func (c ComposeJob) check(id string) error {
|
||||
if c.Package == "" {
|
||||
return fmt.Errorf("buildstep: the compose ask %s names no package to build", id)
|
||||
}
|
||||
if !json.Valid(c.Request) || len(c.Request) == 0 || c.Request[0] != '{' {
|
||||
req, err := c.RequestBytes()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !json.Valid(req) || len(req) == 0 || req[0] != '{' {
|
||||
return fmt.Errorf("buildstep: the compose ask %s holds no request object", id)
|
||||
}
|
||||
return nil
|
||||
|
||||
@@ -144,6 +144,28 @@ func TestAskCheckAndRoundTrip(t *testing.T) {
|
||||
if _, err := DecodeStepAsk([]byte(`{"v":2}`)); !errors.Is(err, ErrUnknownVersion) {
|
||||
t.Errorf("another version: %v", err)
|
||||
}
|
||||
// Issue 1966: a compose job's request travels packed when it passes the bus's message limit, and reads back as sent.
|
||||
t.Run("1966: a compose request travels packed", func(t *testing.T) {
|
||||
obj := []byte(`{"v":1,"input":{"nodes":[]}}`)
|
||||
job := ComposeJob{Fetch: Fetch{Repository: "novox/mesh-host", Commit: strings.Repeat("a", 40)}, Dir: ".", Package: "p"}
|
||||
if err := job.PackRequest(obj); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got, err := job.RequestBytes(); err != nil || string(got) != string(obj) {
|
||||
t.Errorf("read back %s, %v", got, err)
|
||||
}
|
||||
if err := job.check("ask"); err != nil {
|
||||
t.Errorf("a packed request is refused: %v", err)
|
||||
}
|
||||
job.Request = json.RawMessage(`"not packed"`)
|
||||
if err := job.check("ask"); err == nil {
|
||||
t.Errorf("a string that is no packed request is accepted")
|
||||
}
|
||||
job.Request = obj
|
||||
if got, err := job.RequestBytes(); err != nil || string(got) != string(obj) {
|
||||
t.Errorf("a whole request read back %s, %v", got, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// The seat's value and the neutral form are the same JSON: AskOf takes any seat's type (here its shape, with a
|
||||
|
||||
Reference in New Issue
Block a user