diff --git a/modules/brother/README.md b/modules/brother/README.md index ff4449cc..7393401c 100644 --- a/modules/brother/README.md +++ b/modules/brother/README.md @@ -55,7 +55,7 @@ makes it (mode 0755) when it files the first scan. | verb (tool) | does | |---|---| -| `scanner.scan` | `source` (`glass`, default, or `feeder`) and `duplex` (feeder only): asks the device process to scan, crop (glass only), keep the PDF in the bucket under its sha256 and say `scan.added`; then fetches it in parts, checks the sha256 and files it in `scan-folder`. Waits up to 180 s. Answers the event's body and `path` | +| `scanner.scan` | `source` (`glass`, default, or `feeder`) and `duplex` (feeder only): asks the device process to scan, crop (glass only), keep the PDF in the bucket under its sha256 and say `scan.added`; then fetches it in parts, checks the sha256 and files it in `scan-folder`, in the background, so a scan is filed whether or not the caller still waits. A glass scan is waited for up to 180 s. A feeder scan is waited for 25 s (the mesh console gives up after 30 s); if it is not done, the answer is `started: true` with its `id`, and `scanner.scan {id}` reads the outcome (`state` running, done with the event's body and `path`, or failed with `why`). The device process ends a feeder job at 20 s a page (at least 3 min, at most 30 min). A scan whose PDF the tool could not fetch (over 100 MiB) is refused before it is kept or announced. Answers the event's body and `path` | | `scanner.document` | one part of a kept scan (`object`, `part` from 1): `part`, `parts`, `size`, `sha256` (of the whole), `data` (base64). Join the parts and check the sha256 | | `scanner.status` | whether the scanner answers, its state (Idle or Processing) and its document feeder's state; a scanner that does not answer is `answers: false` with why. This is how its health is read: the module has no long-running resource to declare a health check on | | `scanner.list` | the newest filed scans, newest first (`limit`, default 20). Read-only | diff --git a/modules/brother/cmd/brother/deviceproc.go b/modules/brother/cmd/brother/deviceproc.go index 2910c1fc..c101c48f 100644 --- a/modules/brother/cmd/brother/deviceproc.go +++ b/modules/brother/cmd/brother/deviceproc.go @@ -104,7 +104,13 @@ func (w *deviceWorker) Handle(body []byte, early func()) (action, time.Duration) early() settled = true } - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + // A scan of the feeder may run as long as FeederMaxBound (the scan itself ends sooner by FeederBound, which + // scales with its pages) and the PDF is then stored; every other ask is done in 5 minutes. + bound := 5 * time.Minute + if ask.Op == OpScan && ask.Source == "feeder" { + bound = FeederMaxBound + 5*time.Minute + } + ctx, cancel := context.WithTimeout(context.Background(), bound) defer cancel() var answer map[string]any if got, err := w.run(ctx, ask); err != nil { @@ -185,6 +191,11 @@ func (w *deviceWorker) scan(ctx context.Context, ask Ask) (map[string]any, error if err != nil { return nil, err } + // A scan the tool could not take is not kept nor announced: an announced scan is always filed (issue 1539). + if most := MostAnswerParts * AnswerPartSize; len(c.PDF) > most { + return nil, fmt.Errorf("the scan's PDF is %d bytes, over the %d the tool can fetch (%d parts of %d): it is not kept; scan with a lower resolution or fewer sheets", + len(c.PDF), most, MostAnswerParts, AnswerPartSize) + } body := c.Body(w.Node, set.ColorMode) key := body["object"].(string) if err := Store(ctx, objects, key, "application/pdf", c.PDF); err != nil { diff --git a/modules/brother/cmd/brother/escl.go b/modules/brother/cmd/brother/escl.go index a16f1e31..4ae2f7a1 100644 --- a/modules/brother/cmd/brother/escl.go +++ b/modules/brother/cmd/brother/escl.go @@ -128,8 +128,21 @@ type ScanOptions struct { ColorMode string Feeder bool Duplex bool + // Bound is how long a feeder job may have run when it has this many pages and is asked for the next; nil is + // FeederBound. The job is deleted and an error given when it has run longer. + Bound func(pages int) time.Duration } +// FeederBound is how long a feeder job may run when it has this many pages (issue 1539): 20 seconds for each, +// at least 3 minutes and at most 30. The scanner gives a duplex sheet about every 3.4 seconds, so this is generous +// and only ends a job that is stuck. +func FeederBound(pages int) time.Duration { + return min(max(time.Duration(pages)*20*time.Second, 3*time.Minute), FeederMaxBound) +} + +// FeederMaxBound is the longest a feeder job may run. +const FeederMaxBound = 30 * time.Minute + // A4 in the 300ths of an inch the region is asked in; a region is cut to what the input can scan. const a4Width, a4Height = 2480, 3508 @@ -202,7 +215,7 @@ func (c *Client) ScanPages(ctx context.Context, o ScanOptions) ([][]byte, error) } doc := job + "/NextDocument" if o.Feeder { - pages, err := c.feederPages(ctx, doc) + pages, err := c.feederPages(ctx, doc, o.Bound) if err != nil { c.deleteJob(job) return nil, err @@ -226,9 +239,18 @@ func (c *Client) ScanPages(ctx context.Context, o ScanOptions) ([][]byte, error) // feederPages asks NextDocument until the scanner answers 404, which ends the job. The first page missing means no // paper was in the feeder. -func (c *Client) feederPages(ctx context.Context, doc string) ([][]byte, error) { +func (c *Client) feederPages(ctx context.Context, doc string, bound func(int) time.Duration) ([][]byte, error) { + if bound == nil { + bound = FeederBound + } + start := time.Now() var pages [][]byte for len(pages) <= maxFeederPages { + if len(pages) > 0 { + if run, limit := time.Since(start), bound(len(pages)); run > limit { + return nil, fmt.Errorf("the feeder scan has run %s and has %d pages, past its bound of %s: it is stopped", run.Round(time.Second), len(pages), limit) + } + } got, err := c.get(ctx, doc) if err != nil { return nil, fmt.Errorf("the scanned page %d cannot be fetched: %w", len(pages)+1, err) diff --git a/modules/brother/cmd/brother/feeder_test.go b/modules/brother/cmd/brother/feeder_test.go index 66dd63f8..d70ec46a 100644 --- a/modules/brother/cmd/brother/feeder_test.go +++ b/modules/brother/cmd/brother/feeder_test.go @@ -3,9 +3,11 @@ package main import ( "bytes" "context" + "errors" "os" "strings" "testing" + "time" ) // 1452: the Brother ignores the whole body when it holds a line break before the closing root tag, and runs its @@ -163,3 +165,138 @@ func TestDuplexOnTheGlassIsRefusedAndNothingIsAsked(t *testing.T) { t.Errorf("%d asks queued, %d jobs posted", r.bus.asks(), f.posts) } } + +// pdfCapture is a stand-in capture of a small PDF, ready to be released. +func standInCapture(release <-chan struct{}, fail error) func(ctx context.Context, req ScanRequest) (Capture, error) { + return func(ctx context.Context, req ScanRequest) (Capture, error) { + <-release + if fail != nil { + return Capture{}, fail + } + pdf := pdfOf(5000) + return Capture{PDF: pdf, SHA256: sha256Hex(pdf), Resolution: 200, Pages: 3, At: fixedNow}, nil + } +} + +// waitFor polls until the condition holds. +func waitFor(t *testing.T, what string, cond func() bool) { + t.Helper() + for i := 0; i < 400; i++ { + if cond() { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("timed out waiting for %s", what) +} + +// 1552 and 1539: a feeder scan not done within the quick wait is answered as started with its id; the scan goes on +// and is filed although the caller has gone; the id reads the outcome, running and then done with the path. +func TestALongFeederScanAnswersStartedAndIsFiledWithoutTheCaller(t *testing.T) { + f := newFakeScanner(t, fixture(t, "scan.jpg")) + r, folder := scanRig(t, f, Settings{}) + old := quickWait + quickWait = 50 * time.Millisecond + t.Cleanup(func() { quickWait = old }) + release := make(chan struct{}) + r.worker.Capture = standInCapture(release, nil) + + got, err := r.call(t, "scanner.scan", scanArgs("feeder", false)) + if err != nil { + t.Fatal(err) + } + start := got.(map[string]any) + id, _ := start["id"].(string) + if start["started"] != true || id == "" || start["path"] != nil { + t.Fatalf("answer = %v, want started with an id", start) + } + running, err := r.call(t, "scanner.scan", map[string]any{"id": id}) + if err != nil || running.(map[string]any)["state"] != "running" { + t.Fatalf("while it runs: %v, %v", running, err) + } + close(release) + // Nobody asks again: the scan is filed all the same. + waitFor(t, "the scan to be filed", func() bool { e, _ := os.ReadDir(folder); return len(e) == 1 }) + done, err := r.call(t, "scanner.scan", map[string]any{"id": id}) + if err != nil { + t.Fatal(err) + } + d := done.(map[string]any) + if d["state"] != "done" || d["pages"] != float64(3) || d["path"] == nil { + t.Fatalf("outcome = %v, want done with pages and the path", d) + } + if _, err := os.Stat(d["path"].(string)); err != nil { + t.Error(err) + } +} + +// 1552: a scan that fails after the caller was told it started is read as failed, with why; an unknown id is an +// error that says so. +func TestAFailedLongScanIsReadAsFailedWithWhy(t *testing.T) { + f := newFakeScanner(t, fixture(t, "scan.jpg")) + r, _ := scanRig(t, f, Settings{}) + old := quickWait + quickWait = 50 * time.Millisecond + t.Cleanup(func() { quickWait = old }) + release := make(chan struct{}) + r.worker.Capture = standInCapture(release, errors.New("the scanner dropped the job: 410 Gone")) + start, err := r.call(t, "scanner.scan", scanArgs("feeder", true)) + if err != nil { + t.Fatal(err) + } + id := start.(map[string]any)["id"].(string) + close(release) + var out map[string]any + waitFor(t, "the failure", func() bool { + got, _ := r.call(t, "scanner.scan", map[string]any{"id": id}) + out = got.(map[string]any) + return out["state"] == "failed" + }) + if !strings.Contains(out["why"].(string), "410 Gone") { + t.Errorf("why = %v", out["why"]) + } + if _, err := r.call(t, "scanner.scan", map[string]any{"id": "nosuchid"}); err == nil || !strings.Contains(err.Error(), "nosuchid") { + t.Errorf("an unknown id: %v", err) + } +} + +// 1539: the feeder job's bound scales with its pages, from 3 to 30 minutes. +func TestTheFeederBoundScalesWithThePages(t *testing.T) { + for pages, want := range map[int]time.Duration{1: 3 * time.Minute, 9: 3 * time.Minute, 30: 10 * time.Minute, 100: 30 * time.Minute, 500: 30 * time.Minute} { + if got := FeederBound(pages); got != want { + t.Errorf("FeederBound(%d) = %s, want %s", pages, got, want) + } + } + f := newFakeScanner(t, fixture(t, "scan.jpg")) + f.sheets = 3 + _, err := NewClient(f.base()).ScanPages(context.Background(), ScanOptions{Resolution: 200, ColorMode: "color", Feeder: true, + Bound: func(int) time.Duration { return 0 }}) + if err == nil || !strings.Contains(err.Error(), "bound") { + t.Fatalf("error = %v, want the bound", err) + } + if f.open || f.deletes != 1 { + t.Errorf("job open %v, %d DELETEs, want the job deleted", f.open, f.deletes) + } +} + +// 1539: a scan too large for the tool to fetch is not kept nor announced: it is refused with its size. +func TestAScanTooLargeToFetchIsNeitherKeptNorAnnounced(t *testing.T) { + f := newFakeScanner(t, fixture(t, "scan.jpg")) + r, folder := scanRig(t, f, Settings{}) + release := make(chan struct{}) + close(release) + huge := make([]byte, MostAnswerParts*AnswerPartSize+1) + r.worker.Capture = func(ctx context.Context, req ScanRequest) (Capture, error) { + return Capture{PDF: huge, SHA256: "x", Resolution: 200, Pages: 99, At: fixedNow}, nil + } + _, err := r.call(t, "scanner.scan", scanArgs("feeder", false)) + if err == nil || !strings.Contains(err.Error(), "not kept") { + t.Fatalf("error = %v", err) + } + if len(*r.said) != 0 { + t.Errorf("%d events said for a scan not kept", len(*r.said)) + } + if e, _ := os.ReadDir(folder); len(e) != 0 { + t.Errorf("filed: %v", e) + } +} diff --git a/modules/brother/cmd/brother/main.go b/modules/brother/cmd/brother/main.go index 99ec4b1f..436bb9c6 100644 --- a/modules/brother/cmd/brother/main.go +++ b/modules/brother/cmd/brother/main.go @@ -85,13 +85,15 @@ func askFrom(args map[string]any) (Ask, error) { // The settings are read at each call, so a changed config value is the next call's. None opens a credential: what // needs one is asked of the device process. func tools(settingsFile string, now func() time.Time) []stdio.Tool { + jobs := NewJobs() service := func() (*Service, error) { s, err := LoadSettings(settingsFile) if err != nil { return nil, err } return &Service{Settings: s, Node: os.Getenv("MESH_NODE"), Now: now, Asks: askBus, - AnswerWait: answerWait, ScanWait: scanWait, AnswerPoll: answerPoll}, nil + AnswerWait: answerWait, ScanWait: scanWait, AnswerPoll: answerPoll, + QuickWait: quickWait, FeederWait: feederWait, Jobs: jobs}, nil } // ask is a tool that reads its arguments, and answers what the service gives. ask := func(run func(s *Service, ctx context.Context, a Ask) (map[string]any, error), bound time.Duration) func(map[string]any) (any, error) { @@ -217,15 +219,32 @@ func tools(settingsFile string, now func() time.Time) []stdio.Tool { "the platen's white when crop is set) or from the document feeder (source feeder: every sheet in it, whole, up to 50; " + "duplex true scans both sides of each sheet). The device process keeps the result as one PDF in the module's bucket " + "under its sha256 (one page for the glass, one page per side for the feeder) and says scan.added on the bus; this " + - "tool then fetches it and files it in the scan folder. Waits up to three minutes. Answers the event's body: " + + "tool then fetches it and files it in the scan folder, in the background, so a scan is filed even when you stop waiting. " + + "A glass scan is waited for (up to three minutes). A feeder scan is waited for 25 seconds (the mesh console gives up on a " + + "call after 30): when it is not done by then the answer is {started: true, id, state: running}; the scan goes on (the " + + "device process lets a feeder run last 20 seconds a page, 3 to 30 minutes) and is filed when done, and " + + "scanner.scan {id} reads the outcome: state running, state done with the answer below, or state failed with why. " + + "The id is known to the tool process that started the scan; scan.added also announces it. Answers the event's body: " + "the object's key, its size and sha256, the first page's real size in millimetres, the resolution the scanner really gave, " + "whether it was cropped, the number of pages and the file's path. The image itself is never sent. " + "Duplex on the glass is refused.", Input: map[string]any{"type": "object", "properties": map[string]any{ "source": map[string]any{"type": "string", "enum": []string{"glass", "feeder"}, "description": "glass (default) or feeder"}, + "id": map[string]any{"type": "string", "description": "read the outcome of the scan with this id (from an answer that said started) instead of scanning"}, "duplex": map[string]any{"type": "boolean", "description": "feeder only: scan both sides of each sheet (default false)"}, }}, Run: func(args map[string]any) (any, error) { + s, err := service() + if err != nil { + return nil, err + } + id, err := strArg(args, "id") + if err != nil { + return nil, err + } + if id != "" { + return s.ScanResult(id) + } source, err := strArg(args, "source") if err != nil { return nil, err @@ -236,11 +255,7 @@ func tools(settingsFile string, now func() time.Time) []stdio.Tool { return nil, fmt.Errorf("duplex is true or false") } } - s, err := service() - if err != nil { - return nil, err - } - ctx, cancel := context.WithTimeout(context.Background(), scanWait+10*time.Minute) + ctx, cancel := context.WithTimeout(context.Background(), feederWait+10*time.Minute) defer cancel() return s.ScanNow(ctx, source, duplex) }, diff --git a/modules/brother/cmd/brother/printer.go b/modules/brother/cmd/brother/printer.go index d149dcaf..bcf7c8ed 100644 --- a/modules/brother/cmd/brother/printer.go +++ b/modules/brother/cmd/brother/printer.go @@ -39,6 +39,12 @@ const ( // 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 + // QuickWait is how long a feeder scan is waited for in one call: the mesh console gives up on a call after + // 30 seconds, so a longer scan is answered as started, with its id, and read later (issue 1552). + QuickWait = 25 * time.Second + // FeederWait is how long the tool waits in the background for a feeder scan: the longest the device process + // lets one run (FeederMaxBound) and two minutes to store it. + FeederWait = FeederMaxBound + 2*time.Minute // StaleAfter is how old an ask may be when the device process takes it: older than twice the longest wait, // nobody waits for it any more, so the device is not touched for it. StaleAfter = 2 * ScanWait @@ -99,6 +105,8 @@ func (sdkAskBus) Answer(id string) (json.RawMessage, error) { var ( answerWait = AnswerWait scanWait = ScanWait + quickWait = QuickWait + feederWait = FeederWait answerPoll = 500 * time.Millisecond ) @@ -114,6 +122,24 @@ func newAskID() string { // call queues the ask and waits for the device process's answer, at most wait. An answer that is an error is // the error. Nothing here reads, opens or sends a credential (ADR 0436 ยง12). func (s *Service) call(ctx context.Context, ask Ask, wait time.Duration) (map[string]any, error) { + id, err := s.queue(ask) + if err != nil { + return nil, err + } + return s.await(ctx, id, wait) +} + +// queue puts the ask on the work queue and answers its id, under which the answer is written. +func (s *Service) queue(ask Ask) (string, error) { + ask.ID, ask.At = newAskID(), s.Now().UTC() + if err := s.Asks.Publish(ask.ID, ask); err != nil { + return "", fmt.Errorf("the ask for the device process could not be queued: %w", err) + } + return ask.ID, nil +} + +// await waits at most wait for the answer to the ask with this id. An answer that is an error is the error. +func (s *Service) await(ctx context.Context, id string, wait time.Duration) (map[string]any, error) { poll := s.AnswerPoll if wait <= 0 { wait = AnswerWait @@ -121,14 +147,10 @@ func (s *Service) call(ctx context.Context, ask Ask, wait time.Duration) (map[st if poll <= 0 { poll = 500 * time.Millisecond } - ask.ID, ask.At = newAskID(), s.Now().UTC() - if err := s.Asks.Publish(ask.ID, ask); err != nil { - return nil, fmt.Errorf("the ask for the device process could not be queued: %w", err) - } ctx, cancel := context.WithTimeout(ctx, wait) defer cancel() for { - raw, err := s.Asks.Answer(ask.ID) + raw, err := s.Asks.Answer(id) if err != nil { return nil, fmt.Errorf("the device process's answers cannot be read: %w", err) } @@ -145,7 +167,7 @@ func (s *Service) call(ctx context.Context, ask Ask, wait time.Duration) (map[st select { case <-ctx.Done(): return nil, fmt.Errorf("the device process did not answer within %s: it may not be running, or the device is slow (ask %s)", - wait, ask.ID) + wait, id) case <-time.After(poll): } } diff --git a/modules/brother/cmd/brother/scan.go b/modules/brother/cmd/brother/scan.go index 2fadccf7..a224db9e 100644 --- a/modules/brother/cmd/brother/scan.go +++ b/modules/brother/cmd/brother/scan.go @@ -4,6 +4,7 @@ import ( "context" "encoding/base64" "fmt" + "sync" "time" ) @@ -23,23 +24,127 @@ 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 + // 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. + Jobs *Jobs +} + +// Jobs are the scans this tool process has started and not yet been asked the outcome of. A scan runs in the +// background from the moment it is asked for, so that it is filed whether or not the caller still waits (issue +// 1539); the caller waits for it, and reads its outcome later with its id when it did not finish in time. +type Jobs struct { + mu sync.Mutex + m map[string]*job +} + +type job struct { + done chan struct{} + body map[string]any // set when done + err error // set when failed +} + +// NewJobs is an empty set of jobs. +func NewJobs() *Jobs { return &Jobs{m: map[string]*job{}} } + +func (j *Jobs) get(id string) *job { + if j == nil { + return nil + } + j.mu.Lock() + defer j.mu.Unlock() + return j.m[id] +} + +func (j *Jobs) put(id string) *job { + j.mu.Lock() + defer j.mu.Unlock() + if j.m == nil { + j.m = map[string]*job{} + } + x := &job{done: make(chan struct{})} + j.m[id] = x + return x } // ScanNow asks the device process to scan, crop, build the PDF, keep it in the bucket and say scan.added, waits for // the stored scan's key, fetches its parts through the device process, joins them, checks their SHA-256 and files // the PDF in the scan folder (temp and rename, no overwrite). It answers the event's body and the path. Nothing is // asked before scanner and scan-folder are known to be set (1440/AC7); the tool never reads a credential. +// +// The scan is followed by a goroutine of its own to the end, so it is filed even when the caller has gone. A glass +// scan is waited for (ScanWait); a feeder scan only for QuickWait, and when it is not finished by then the answer is +// started: true with the scan's id, and ScanResult reads the outcome later. func (s *Service) ScanNow(ctx context.Context, source string, duplex bool) (map[string]any, error) { if err := s.Settings.Missing(); err != nil { return nil, err } - if _, err := ParseSource(source, duplex); err != nil { - return nil, err - } - ans, err := s.call(ctx, Ask{Op: OpScan, Source: source, Duplex: duplex}, s.ScanWait) + req, err := ParseSource(source, duplex) if err != nil { return nil, err } + id, err := s.queue(Ask{Op: OpScan, Source: source, Duplex: duplex}) + if err != nil { + return nil, err + } + bound, quick := s.ScanWait, s.ScanWait + if req.Feeder { + bound, quick = s.FeederWait, s.QuickWait + } + j := s.Jobs.put(id) + go func() { + // The goroutine outlives the caller: a context of its own, bounded by the scan's own bound and the time + // to fetch the parts. + ctx, cancel := context.WithTimeout(context.Background(), bound+10*time.Minute) + defer cancel() + j.body, j.err = s.finishScan(ctx, id, bound) + close(j.done) + }() + select { + case <-j.done: + return j.body, j.err + case <-time.After(quick): + return map[string]any{"started": true, "id": id, "state": "running", + "note": "the scan goes on and is filed when it is done; read its outcome with scanner.scan {id}, or wait for scan.added"}, nil + case <-ctx.Done(): + return nil, fmt.Errorf("the caller went away; the scan %s goes on and is filed: %w", id, ctx.Err()) + } +} + +// ScanResult is the outcome of the scan with this id: its body and path when it is done, state running when it is +// not, and state failed with why when it failed. A scan this process does not know (it was restarted, or the id is +// wrong) is an error. +func (s *Service) ScanResult(id string) (map[string]any, error) { + j := s.Jobs.get(id) + if j == nil { + return nil, fmt.Errorf("no scan %s is known to this tool: its id is wrong, or the tool was restarted since it was asked for (a scan the device process kept is announced by scan.added)", id) + } + select { + case <-j.done: + if j.err != nil { + return map[string]any{"id": id, "state": "failed", "why": j.err.Error()}, nil + } + out := map[string]any{"id": id, "state": "done"} + for k, v := range j.body { + out[k] = v + } + return out, nil + default: + return map[string]any{"started": true, "id": id, "state": "running"}, nil + } +} + +// finishScan waits for the device process's answer to the scan ask, then fetches the stored scan and files it. +func (s *Service) finishScan(ctx context.Context, id string, wait time.Duration) (map[string]any, error) { + ans, err := s.await(ctx, id, wait) + if err != nil { + return nil, err + } + return s.fileAnswer(ctx, ans) +} + +// fileAnswer fetches the scan the device process's answer names and files it. +func (s *Service) fileAnswer(ctx context.Context, ans map[string]any) (map[string]any, error) { key, _ := ans["key"].(string) sum, _ := ans["sha256"].(string) parts, _ := ans["parts"].(float64)