Answer a long feeder scan as started with an id, file it in the background, and scale the feeder's bound with its pages (issues 1552, 1539)

This commit is contained in:
2026-10-21 01:40:29 +02:00
parent 7bf63f9eff
commit 23c3a2396e
7 changed files with 333 additions and 21 deletions
+1 -1
View File
@@ -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 |
+12 -1
View File
@@ -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 {
+24 -2
View File
@@ -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)
+137
View File
@@ -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)
}
}
+22 -7
View File
@@ -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)
},
+28 -6
View File
@@ -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):
}
}
+109 -4
View File
@@ -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)