diff --git a/modules/brother/cmd/brother/documents_test.go b/modules/brother/cmd/brother/documents_test.go index 87e0d014..e645a02d 100644 --- a/modules/brother/cmd/brother/documents_test.go +++ b/modules/brother/cmd/brother/documents_test.go @@ -271,6 +271,8 @@ type ippFake struct { // formats is document-format-supported, as the printer lists it; asked counts the Get-Printer-Attributes requests. formats []string asked int + // delay is how long a Print-Job takes to be answered. + delay time.Duration } func newIPPFake(t *testing.T, jobID, jobState int) *ippFake { @@ -298,6 +300,7 @@ func newIPPFake(t *testing.T, jobID, jobState int) *ippFake { return } f.body = body + time.Sleep(f.delay) ans := []byte{0x02, 0x00, byte(f.status >> 8), byte(f.status), 0, 0, 0, 1, ippOperation} ans = appendAttr(ans, ippCharset, "attributes-charset", "utf-8") ans = append(ans, ippJob) diff --git a/modules/brother/cmd/brother/main.go b/modules/brother/cmd/brother/main.go index 0b0f1bf2..1b4209e9 100644 --- a/modules/brother/cmd/brother/main.go +++ b/modules/brother/cmd/brother/main.go @@ -87,7 +87,7 @@ func askFrom(args map[string]any) (Ask, error) { var a Ask var err error for name, dst := range map[string]*string{"page": &a.Page, "object": &a.Object, "upload": &a.Upload, "sha256": &a.SHA256, - "data": &a.Data, "sides": &a.Sides, "media": &a.Media, "path": &a.Path, "node": &a.Node, "url": &a.URL} { + "data": &a.Data, "sides": &a.Sides, "media": &a.Media, "path": &a.Path, "node": &a.Node, "url": &a.URL, "ask": &a.Of} { if *dst, err = strArg(args, name); err != nil { return a, err } @@ -111,7 +111,7 @@ func tools(settingsFile string, now func() time.Time) []stdio.Tool { return nil, err } return &Service{Settings: s, Node: os.Getenv("MESH_NODE"), Now: now, Asks: askBus, - AnswerWait: answerWait, ScanWait: scanWait, AnswerPoll: answerPoll, PrintWait: PrintWait, + AnswerWait: answerWait, ScanWait: scanWait, AnswerPoll: answerPoll, QuickWait: quickWait, FeederWait: feederWait, Jobs: jobs, Upload: uploadVia, Bucket: func() (string, error) { return ReadBucket(os.Getenv("BROTHER_BUCKET")) }}, nil } @@ -156,12 +156,13 @@ func tools(settingsFile string, now func() time.Time) []stdio.Tool { "refused there by name and nothing is printed), or a web link (url). Name exactly one. A Word, spreadsheet or slide " + "document is turned into a PDF first; otherwise only PDF and JPEG, told by the bytes and not the name, and only a format " + "the printer lists in document-format-supported. The device process sends it as an IPP Print-Job and answers the job's " + - "id and state. No document crosses the bus.", + "id and state. No document crosses the bus. A call answers within 25 s (the tool runner cuts a call at 30 s): a print not done by then goes on and is answered as started with its ask id, and printer.print with only that ask reads it.", Input: map[string]any{"type": "object", "properties": map[string]any{ "object": str("a stored document's key: a scan's key, as scan.added names it, or the key node-files.upload answered"), "path": str("an absolute path of a file on the machine node names"), "node": str("the machine path is on, such as g14"), "url": str("an http or https address of a document or a page"), + "ask": str("the ask id of an earlier print that was answered as started: alone, it reads that print's outcome"), "copies": num("1 to 99 (default 1)"), "sides": str("one-sided (default), two-sided-long-edge or two-sided-short-edge"), "media": str("the paper's size name (default " + DefaultMedia + ")"), @@ -171,6 +172,9 @@ func tools(settingsFile string, now func() time.Time) []stdio.Tool { if err != nil { return nil, err } + if a.Of != "" && len(args) > 1 { + return nil, fmt.Errorf("ask is read on its own: give the ask id alone to read an earlier print") + } s, err := service() if err != nil { return nil, err diff --git a/modules/brother/cmd/brother/printer.go b/modules/brother/cmd/brother/printer.go index d2b33511..75cba5ee 100644 --- a/modules/brother/cmd/brother/printer.go +++ b/modules/brother/cmd/brother/printer.go @@ -36,8 +36,11 @@ const ( // DeviceProcessID is the device process's id as module.json's resource "device" names it: the controller // delivers its bus account as the own secret broker-device. DeviceProcessID = "device" - // PrintWait is how long printer.print waits for the device process: a fetch and a conversion come before the job. - PrintWait = 4 * time.Minute + // PrintTakeUp is how long a print ask may wait on the queue before the device process takes it up, and PrintFollow + // how long the tool side follows it to its end: the device process may fetch and convert before the job. A call + // answers within QuickWait whatever these are, and the print goes on (the tool runner cuts every call at 30 s). + PrintTakeUp = 3 * time.Minute + PrintFollow = 6 * time.Minute // AnswerWait is how long a verb waits for the device process; ScanWait how long a scan does. AnswerWait = 60 * time.Second ScanWait = 180 * time.Second @@ -95,6 +98,7 @@ type Ask struct { // before anything is queued, so the queued ask carries a key and the options and never these (ADR 0468 clause 5). // A URL is queued as given: the device process alone fetches it (clause 6). Path string `json:"-"` // print: a file on the machine Node + Of string `json:"-"` // print: the ask id of an earlier print, whose outcome is read Node string `json:"-"` // print: the machine Path is on URL string `json:"url,omitempty"` // print: a web document the device process fetches (issue 1599) } @@ -154,7 +158,10 @@ func (s *Service) call(ctx context.Context, ask Ask, wait time.Duration) (map[st // queue puts the ask on the work queue and answers its id, under which the answer is written. The asker waits // startBy for the device process to take the ask up (0 is for as long as it is waited for). func (s *Service) queue(ask Ask, startBy time.Duration) (string, error) { - ask.ID, ask.At = newAskID(), s.Now().UTC() + if ask.ID == "" { + ask.ID = newAskID() + } + ask.At = s.Now().UTC() if startBy > 0 { ask.Deadline = ask.At.Add(startBy) } @@ -354,7 +361,16 @@ func sdkUpload(node, path, bucket string) (string, error) { // PrinterPrint prints a stored document, a file of a machine, or a web link, in one call. A file of a machine is first // sent into the bucket by that machine's node-files (refused there by name when it is a secret, and then nothing is // printed); the print ask the device process takes carries the key and the options and no bytes. +// +// The tool runner cuts every call at 30 s, so the call answers within QuickWait, upload included. A print not done by +// then goes on in the background and is answered as started, with its ask id; printer.print {ask} reads its outcome. func (s *Service) PrinterPrint(ctx context.Context, a Ask) (map[string]any, error) { + if a.Of != "" { + if a.Object != "" || a.Path != "" || a.Node != "" || a.URL != "" || a.Upload != "" { + return nil, fmt.Errorf("ask is read on its own: name exactly one of object, path with node, url, or the ask id of an earlier print") + } + return s.PrintResult(ctx, a.Of) + } a.Op = OpPrint if err := checkPrint(&a); err != nil { return nil, err @@ -362,7 +378,7 @@ func (s *Service) PrinterPrint(ctx context.Context, a Ask) (map[string]any, erro if s.Settings.Printer == "" { return nil, unsetError("printer") } - from := map[string]any{} + bucket := "" switch { case a.URL != "": if err := CheckURLScheme(a.URL); err != nil { @@ -372,10 +388,24 @@ func (s *Service) PrinterPrint(ctx context.Context, a Ask) (map[string]any, erro if s.Upload == nil { return nil, fmt.Errorf("this module has no way to ask node-files for an upload") } - bucket, err := s.Bucket() - if err != nil { + var err error + if bucket, err = s.Bucket(); err != nil { return nil, err } + } + id := newAskID() + j := s.Jobs.put(id) + go func() { + j.body, j.err = s.printRun(id, a, bucket) + close(j.done) + }() + return s.printWait(ctx, id, j) +} + +// printRun is a print followed to its end, however long the caller waits: the upload, the ask, the device's answer. +func (s *Service) printRun(id string, a Ask, bucket string) (map[string]any, error) { + from := map[string]any{} + if a.Path != "" { key, err := s.Upload(a.Node, a.Path, bucket) if err != nil { return nil, fmt.Errorf("%s on %s could not be sent to the bucket, and nothing is printed: %w", a.Path, a.Node, err) @@ -383,8 +413,13 @@ func (s *Service) PrinterPrint(ctx context.Context, a Ask) (map[string]any, erro from["path"], from["node"] = a.Path, a.Node a.Object, a.Path, a.Node = key, "", "" } - wait := max(s.AnswerWait, s.PrintWait) - out, err := s.call(ctx, a, wait) + a.ID = id + if _, err := s.queue(a, PrintTakeUp); err != nil { + return nil, err + } + ctx, cancel := context.WithTimeout(context.Background(), PrintFollow) + defer cancel() + out, err := s.await(ctx, id, PrintFollow) if err != nil { return nil, err } @@ -394,6 +429,52 @@ func (s *Service) PrinterPrint(ctx context.Context, a Ask) (map[string]any, erro return out, nil } +// printWait gives the print's outcome if it comes within QuickWait, and started with the ask id otherwise. +func (s *Service) printWait(ctx context.Context, id string, j *job) (map[string]any, error) { + quick := s.QuickWait + if quick <= 0 { + quick = QuickWait + } + select { + case <-j.done: + return j.body, j.err + case <-time.After(quick): + return map[string]any{"state": "started", "ask": id, + "note": "the print goes on; read its outcome with printer.print {ask}"}, nil + case <-ctx.Done(): + return nil, fmt.Errorf("the caller went away; the print %s goes on: %w", id, ctx.Err()) + } +} + +// PrintResult is the outcome of the print with this ask id: the device's answer when it is done (a failure is the +// error), and started, still printing, while it is not. A print this process followed is read from it, one it did not +// (it was restarted) from the device's answers, which are kept for 10 minutes. +func (s *Service) PrintResult(ctx context.Context, id string) (map[string]any, error) { + if !askIDPattern.MatchString(id) { + return nil, fmt.Errorf("ask %q is not an ask id: the 24 hex digits printer.print answered", id) + } + if j := s.Jobs.get(id); j != nil { + return s.printWait(ctx, id, j) + } + raw, err := s.Asks.Answer(id) + if err != nil { + return nil, fmt.Errorf("the device process's answers cannot be read: %w", err) + } + if raw == nil { + return map[string]any{"state": "started", "ask": id, "note": "still printing, or this ask is not known"}, nil + } + var out map[string]any + if err := json.Unmarshal(raw, &out); err != nil { + return nil, fmt.Errorf("the device process's answer is not JSON: %w", err) + } + if msg, bad := out["error"].(string); bad { + return nil, fmt.Errorf("%s", msg) + } + return out, nil +} + +var askIDPattern = regexp.MustCompile(`^[0-9a-f]{24}$`) + // PrinterStatus is the printer's state over IPP and the scanner's over eSCL, none of it behind a login // (ADR 0467 clause 5). A part that does not answer is an answer, with why, not an error. Which of printer and // scanner is not set is said by name, and nothing is contacted. diff --git a/modules/brother/cmd/brother/printlink_test.go b/modules/brother/cmd/brother/printlink_test.go index 6c532e3e..045c4eb8 100644 --- a/modules/brother/cmd/brother/printlink_test.go +++ b/modules/brother/cmd/brother/printlink_test.go @@ -7,6 +7,7 @@ import ( "path/filepath" "strings" "testing" + "time" ) // uploadFake stands for node-files.upload on another machine: it records the ask and answers a key, or refuses. @@ -184,3 +185,123 @@ func TestManifestSharesPutWithNodeFilesAndInvokesItsUpload(t *testing.T) { } } } + +// quickly makes a call wait this long before it answers started, for the test. +func quickly(t *testing.T, d time.Duration) { + t.Helper() + old := quickWait + quickWait = d + t.Cleanup(func() { quickWait = old }) +} + +// 1598/AC7: a device answer slower than the call's bound is answered as started with the ask id within the bound, and +// printer.print {ask} then returns the device's answer. +func TestASlowPrintIsAnsweredAsStartedAndReadWithItsAsk(t *testing.T) { + fake := newIPPFake(t, 71, 5) + fake.delay = 400 * time.Millisecond + quickly(t, 60*time.Millisecond) + r := newRig(t, `{"printer":"192.0.2.10"}`, "pw") + doc := pdfOf(300) + key := sha256Hex(doc) + ".pdf" + r.objects.Put(t.Context(), key, "application/pdf", doc) + began := time.Now() + got, err := r.call(t, "printer.print", map[string]any{"object": key}) + if err != nil { + t.Fatal(err) + } + m := got.(map[string]any) + id, _ := m["ask"].(string) + if m["state"] != "started" || id == "" || time.Since(began) > 300*time.Millisecond { + t.Fatalf("answer %v after %s", m, time.Since(began)) + } + if QuickWait != 25*time.Second { + t.Errorf("a call waits %s, want 25s", QuickWait) + } + // Read while it goes on, and after: the same id. + deadline := time.Now().Add(5 * time.Second) + for { + read, err := r.call(t, "printer.print", map[string]any{"ask": id}) + if err != nil { + t.Fatal(err) + } + rm := read.(map[string]any) + if rm["state"] == nil { + if rm["job-id"] != float64(71) || rm["object"] != key { + t.Errorf("the outcome = %v", rm) + } + break + } + if time.Now().After(deadline) { + t.Fatalf("never done: %v", rm) + } + } + // A restarted tool process has no job: the device's answers are read. + r2 := newRig(t, `{"printer":"192.0.2.10"}`, "pw") + r2.answers.m = map[string]json.RawMessage{"0123456789abcdef01234567": json.RawMessage(`{"job-id": 5}`)} + if read, err := r2.call(t, "printer.print", map[string]any{"ask": "0123456789abcdef01234567"}); err != nil || read.(map[string]any)["job-id"] != float64(5) { + t.Errorf("read from the answers: %v, %v", read, err) + } + if _, err := r.call(t, "printer.print", map[string]any{"ask": id, "copies": float64(2)}); err == nil { + t.Error("ask with another argument was accepted") + } + if _, err := r.call(t, "printer.print", map[string]any{"ask": "../x"}); err == nil { + t.Error("a made-up ask id was accepted") + } +} + +// 1598/AC8: printer.print {path, node} whose upload and print together pass the bound answers started with the ask id, +// never a timeout, and the print follows. +func TestASlowUploadAndPrintAnswersStartedNeverATimeout(t *testing.T) { + fake := newIPPFake(t, 72, 5) + fake.delay = 100 * time.Millisecond + quickly(t, 60*time.Millisecond) + doc := pdfOf(300) + key := sha256Hex(doc) + up := withUpload(t, key, nil) + old := uploadVia + uploadVia = func(node, path, bucket string) (string, error) { + time.Sleep(150 * time.Millisecond) // node-files is slow + return old(node, path, bucket) + } + t.Cleanup(func() { uploadVia = old }) + r := newRig(t, `{"printer":"192.0.2.10"}`, "pw") + r.objects.Put(t.Context(), key, "application/pdf", doc) + got, err := r.call(t, "printer.print", map[string]any{"path": "/home/ops/letter.pdf", "node": "g14"}) + if err != nil { + t.Fatalf("a slow upload failed the call: %v", err) + } + m := got.(map[string]any) + id, _ := m["ask"].(string) + if m["state"] != "started" || id == "" || len(up.calls) > 1 { + t.Fatalf("answer %v", m) + } + for deadline := time.Now().Add(5 * time.Second); ; { + read, err := r.call(t, "printer.print", map[string]any{"ask": id}) + if err != nil { + t.Fatal(err) + } + if rm := read.(map[string]any); rm["state"] == nil { + if rm["job-id"] != float64(72) || rm["path"] != "/home/ops/letter.pdf" || rm["node"] != "g14" { + t.Errorf("the outcome = %v", rm) + } + break + } + if time.Now().After(deadline) { + t.Fatal("never done") + } + } + if len(up.calls) != 1 || len(r.bus.published) != 1 || r.bus.published[0].Object != key { + t.Errorf("uploads %v, asks %v", up.calls, r.bus.published) + } + // An upload that fails after the call answered is the outcome, read with the ask. + uploadVia = func(node, path, bucket string) (string, error) { + time.Sleep(150 * time.Millisecond) + return "", errors.New("/home/ops/.ssh/id_ed25519 is refused by name") + } + got, _ = r.call(t, "printer.print", map[string]any{"path": "/home/ops/.ssh/id_ed25519", "node": "g14"}) + id2, _ := got.(map[string]any)["ask"].(string) + time.Sleep(300 * time.Millisecond) + if _, err := r.call(t, "printer.print", map[string]any{"ask": id2}); err == nil || !strings.Contains(err.Error(), "refused by name") { + t.Errorf("the failed upload's outcome: %v", err) + } +} diff --git a/modules/brother/cmd/brother/scan.go b/modules/brother/cmd/brother/scan.go index 0475c433..e8ddf14b 100644 --- a/modules/brother/cmd/brother/scan.go +++ b/modules/brother/cmd/brother/scan.go @@ -28,8 +28,6 @@ type Service struct { // AnswerWait is how long a verb waits for the device process's answer (ScanWait for a scan); AnswerPoll how // often it looks. AnswerWait, ScanWait, AnswerPoll time.Duration - // PrintWait is how long printer.print waits: a fetch and a conversion come before the job. - PrintWait time.Duration // QuickWait is how long a feeder scan is waited for in one call, FeederWait how long it is followed. QuickWait, FeederWait time.Duration // Jobs are the scans this process follows.