Merge pull request 'Close every eSCL scan job brother opens, and say a busy scanner as busy (issue 1531)' (#392) from fix/1531-close-scan-jobs into main

modules/brother/cmd/brother/escl.go

@@ -1,27 +1,30 @@
 package main
 
 import (
-	"bytes"
 	"context"
 	"encoding/xml"
 	"fmt"
 	"io"
 	"net/http"
 	"net/url"
+	"os"
 	"strings"
+	"sync"
 	"time"
 )
 
 // eSCL (the AirScan protocol) over HTTP: POST <base>/ScanJobs answers 201 and the job's Location, GET
-// <Location>/NextDocument answers the image, and a second NextDocument answers 404 once the job is done.
+// <Location>/NextDocument answers the image, and a second NextDocument answers 404 once the job is done (that is what closes it; DELETE of the job does too).
 
 // maxImage bounds what a scanner may send: a 600 dpi A4 JPEG is a few tens of megabytes.
 const maxImage = 256 << 20
 
 // Client speaks eSCL to one scanner.
 type Client struct {
-	Base string
-	HTTP *http.Client
+	Base      string
+	HTTP      *http.Client
+	RetryWait time.Duration // the wait before the one retry of a busy scanner (default 4 s)
+	Logf      func(format string, a ...any)
 }
 
 // NewClient is a client for the eSCL base URL, with a timeout that outlasts a slow scan.
@@ -124,6 +127,10 @@ func ScanSettings(c Capabilities, resolution int, colorMode string) (string, err
 
 // Scan puts one job to the scanner and returns the JPEG it sends. The resolution asked is only a request: the
 // real one is the image's own density (Density), for the real scanner answered 300 with 200.
+//
+// A job stays open on the scanner (pwg:State Processing, and 503 to the next ScanJobs) until NextDocument has
+// answered 404 or the job is DELETEd, so Scan always ends the job it made: after the page it asks NextDocument
+// again, and on any failure after the job was created it deletes the job.
 func (c *Client) Scan(ctx context.Context, resolution int, colorMode string) ([]byte, error) {
 	caps, err := c.Capabilities(ctx)
 	if err != nil {
@@ -133,29 +140,99 @@ func (c *Client) Scan(ctx context.Context, resolution int, colorMode string) ([]
 	if err != nil {
 		return nil, err
 	}
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.Base+"/ScanJobs", bytes.NewReader([]byte(settings)))
+	job, err := c.createJob(ctx, settings)
 	if err != nil {
 		return nil, err
 	}
-	req.Header.Set("Content-Type", "text/xml")
-	resp, err := c.HTTP.Do(req)
+	doc := job + "/NextDocument"
+	image, err := c.page(ctx, doc)
 	if err != nil {
-		return nil, fmt.Errorf("the scanner does not answer: %w", err)
+		c.deleteJob(job)
+		return nil, err
+	}
+	closed, extra := c.closeJob(ctx, doc)
+	if !closed {
+		c.deleteJob(job)
+	}
+	if extra > 0 {
+		return nil, fmt.Errorf("the scanner gave more pages than the glass holds: %d", 1+extra)
+	}
+	return image, nil
+}
+
+// leftOpen is the job URL, per scanner, that brother could neither finish nor delete.
+var leftOpen sync.Map
+
+func (c *Client) logf(format string, a ...any) {
+	if c.Logf != nil {
+		c.Logf(format, a...)
+		return
+	}
+	fmt.Fprintf(os.Stderr, format+"\n", a...)
+}
+
+func (c *Client) retryWait() time.Duration {
+	if c.RetryWait > 0 {
+		return c.RetryWait
+	}
+	return 4 * time.Second
+}
+
+// createJob posts the settings and returns the job's URL. A 503 is a scanner busy with another job; when the job
+// brother itself left open on this scanner is now closed, the post goes once more, never more than once.
+func (c *Client) createJob(ctx context.Context, settings string) (string, error) {
+	resp, reply, err := c.postJob(ctx, settings)
+	if err != nil {
+		return "", err
+	}
+	if resp.StatusCode == http.StatusServiceUnavailable {
+		if left, ok := leftOpen.Load(c.Base); ok {
+			select {
+			case <-time.After(c.retryWait()):
+			case <-ctx.Done():
+				return "", ctx.Err()
+			}
+			if c.deleteJob(left.(string)) {
+				if resp, reply, err = c.postJob(ctx, settings); err != nil {
+					return "", err
+				}
+			}
+		}
+		if resp.StatusCode == http.StatusServiceUnavailable {
+			return "", fmt.Errorf("the scanner is busy with another job: %s %s", resp.Status, strings.TrimSpace(string(reply)))
+		}
 	}
-	reply, _ := readBody(resp, 1<<20)
 	if resp.StatusCode != http.StatusCreated {
-		return nil, fmt.Errorf("the scanner refused the scan job: %s %s", resp.Status, strings.TrimSpace(string(reply)))
+		return "", fmt.Errorf("the scanner refused the scan job: %s %s", resp.Status, strings.TrimSpace(string(reply)))
 	}
 	loc := resp.Header.Get("Location")
 	if loc == "" {
-		return nil, fmt.Errorf("the scanner took the scan job but named no Location")
+		return "", fmt.Errorf("the scanner took the scan job but named no Location")
 	}
 	base, _ := url.Parse(c.Base + "/")
 	ref, err := url.Parse(loc)
 	if err != nil {
-		return nil, fmt.Errorf("the scan job's Location %q is not a URL: %w", loc, err)
+		return "", fmt.Errorf("the scan job's Location %q is not a URL: %w", loc, err)
 	}
-	doc := strings.TrimRight(base.ResolveReference(ref).String(), "/") + "/NextDocument"
+	return strings.TrimRight(base.ResolveReference(ref).String(), "/"), nil
+}
+
+func (c *Client) postJob(ctx context.Context, settings string) (*http.Response, []byte, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.Base+"/ScanJobs", strings.NewReader(settings))
+	if err != nil {
+		return nil, nil, err
+	}
+	req.Header.Set("Content-Type", "text/xml")
+	resp, err := c.HTTP.Do(req)
+	if err != nil {
+		return nil, nil, fmt.Errorf("the scanner does not answer: %w", err)
+	}
+	reply, _ := readBody(resp, 1<<20)
+	return resp, reply, nil
+}
+
+// page fetches the job's one page.
+func (c *Client) page(ctx context.Context, doc string) ([]byte, error) {
 	got, err := c.get(ctx, doc)
 	if err != nil {
 		return nil, fmt.Errorf("the scanned page cannot be fetched: %w", err)
@@ -164,6 +241,9 @@ func (c *Client) Scan(ctx context.Context, resolution int, colorMode string) ([]
 	if err != nil {
 		return nil, fmt.Errorf("the scanned page cannot be read: %w", err)
 	}
+	if got.StatusCode == http.StatusGone {
+		return nil, fmt.Errorf("the scanner dropped the job: %s", got.Status)
+	}
 	if got.StatusCode != http.StatusOK {
 		return nil, fmt.Errorf("the scanner gave no page: %s", got.Status)
 	}
@@ -173,6 +253,57 @@ func (c *Client) Scan(ctx context.Context, resolution int, colorMode string) ([]
 	return image, nil
 }
 
+// maxClosing bounds the NextDocument requests that ask for the end of a job.
+const maxClosing = 1
+
+// closeJob asks NextDocument until the scanner answers 404, which ends the job; true when it did (or the job is
+// gone already), and how many further pages it sent instead; any is an error for the caller, for a Platen job holds
+// one page. It stops at the first further page, and the job is then deleted.
+func (c *Client) closeJob(ctx context.Context, doc string) (closed bool, extra int) {
+	ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 15*time.Second)
+	defer cancel()
+	for i := 0; i < maxClosing; i++ {
+		resp, err := c.get(ctx, doc)
+		if err != nil {
+			c.logf("brother: closing the scan job: %v", err)
+			return false, 0
+		}
+		readBody(resp, maxImage)
+		switch resp.StatusCode {
+		case http.StatusNotFound, http.StatusGone:
+			return true, 0
+		case http.StatusOK:
+			c.logf("brother: the scanner sent a further page after the glass's one")
+			return false, 1
+		}
+		c.logf("brother: closing the scan job: NextDocument answered %s", resp.Status)
+		return false, 0
+	}
+	return false, 0
+}
+
+// deleteJob deletes the job, best effort; true when the job is gone. A job that cannot be deleted is remembered, so
+// a 503 on the next scan can be told from another client's job.
+func (c *Client) deleteJob(job string) bool {
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, http.MethodDelete, job, nil)
+	if err == nil {
+		var resp *http.Response
+		if resp, err = c.HTTP.Do(req); err == nil {
+			readBody(resp, 1<<20)
+			if resp.StatusCode/100 == 2 || resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone {
+				leftOpen.Delete(c.Base)
+				return true
+			}
+			err = fmt.Errorf("answered %s", resp.Status)
+		}
+	}
+	c.logf("brother: deleting the scan job %s: %v", job, err)
+	leftOpen.Store(c.Base, job)
+	return false
+}
+
 // Density is the image's own resolution in dpi, from its JFIF header (units dpi, or dots per cm).
 func Density(jpegBytes []byte) (int, error) {
 	b := jpegBytes

modules/brother/cmd/brother/escl_test.go

@@ -4,12 +4,13 @@ import (
 	"context"
 	"strings"
 	"testing"
+	"time"
 )
 
 func squeeze(s string) string { return strings.ReplaceAll(s, "\n", "") }
 
 // 1440/AC1: the captured capabilities, job creation, document and status replies, served by a test server.
-func TestScanPostsOnePlatenJobOverTheWholeGlassAndFetchesTheDocumentOnce(t *testing.T) {
+func TestScanPostsOnePlatenJobOverTheWholeGlassAndAsksNextDocumentAgainToCloseIt(t *testing.T) {
 	f := newFakeScanner(t, fixture(t, "scan.jpg"))
 	image, err := NewClient(f.base()).Scan(context.Background(), 300, "color")
 	if err != nil {
@@ -23,8 +24,9 @@ func TestScanPostsOnePlatenJobOverTheWholeGlassAndFetchesTheDocumentOnce(t *test
 	if got, want := squeeze(f.postBody), squeeze(string(fixture(t, "settings.xml"))); got != want {
 		t.Fatalf("the job is not the captured request\n got: %s\nwant: %s", got, want)
 	}
-	if f.documents != 1 {
-		t.Fatalf("%d NextDocument requests, want exactly 1", f.documents)
+	// The page, then the second request that the scanner answers 404, which closes the job.
+	if f.documents != 2 || f.open || f.deletes != 0 {
+		t.Fatalf("%d NextDocument requests, job open %v, %d DELETEs, want 2, closed, 0", f.documents, f.open, f.deletes)
 	}
 	if want := fixture(t, "scan.jpg"); string(image) != string(want) {
 		t.Fatalf("the returned image is %d bytes, want the scanner's %d", len(image), len(want))
@@ -104,3 +106,113 @@ func TestMissingNamesWhatIsUnset(t *testing.T) {
 		t.Errorf("both set: %v", err)
 	}
 }
+
+// 1531: two scans in a row both succeed, for the scanner answers 503 to a new job while one is open.
+func TestTwoScansInARowBothSucceed(t *testing.T) {
+	f := newFakeScanner(t, fixture(t, "scan.jpg"))
+	c := NewClient(f.base())
+	for i := 1; i <= 2; i++ {
+		if _, err := c.Scan(context.Background(), 300, "color"); err != nil {
+			t.Fatalf("scan %d: %v", i, err)
+		}
+	}
+	if f.posts != 2 || f.open {
+		t.Fatalf("%d posts, job open %v, want 2, closed", f.posts, f.open)
+	}
+}
+
+// 1531: a page that cannot be had (the answer is lost) leaves no job open: the job is DELETEd.
+func TestAJobWhoseNextDocumentFailsIsDeleted(t *testing.T) {
+	f := newFakeScanner(t, fixture(t, "scan.jpg"))
+	f.dropConn = true
+	if _, err := NewClient(f.base()).Scan(context.Background(), 300, "color"); err == nil {
+		t.Fatal("a lost answer is a scan")
+	}
+	if f.deletes != 1 || f.open {
+		t.Fatalf("%d DELETEs, job open %v, want 1, closed", f.deletes, f.open)
+	}
+	f.mu.Lock()
+	f.dropConn = false
+	f.mu.Unlock()
+	if _, err := NewClient(f.base()).Scan(context.Background(), 300, "color"); err != nil {
+		t.Fatalf("the scan after it: %v", err)
+	}
+}
+
+// 1531: a job with no page (404) is deleted too.
+func TestAJobThatGivesNoPageIsDeleted(t *testing.T) {
+	f := newFakeScanner(t, fixture(t, "scan.jpg"))
+	f.noPage = true
+	if _, err := NewClient(f.base()).Scan(context.Background(), 300, "color"); err == nil {
+		t.Fatal("no page is a scan")
+	}
+	if f.open {
+		t.Fatal("the job stays open")
+	}
+}
+
+// 1531: a 410 on NextDocument is said as a dropped job, and the job is deleted.
+func TestAGoneJobIsSaidAsDroppedAndDeleted(t *testing.T) {
+	f := newFakeScanner(t, fixture(t, "scan.jpg"))
+	f.gone = true
+	_, err := NewClient(f.base()).Scan(context.Background(), 300, "color")
+	if err == nil || !strings.Contains(err.Error(), "the scanner dropped the job") {
+		t.Fatalf("error = %v, want the scanner dropped the job", err)
+	}
+	if f.deletes != 1 {
+		t.Fatalf("%d DELETEs, want 1", f.deletes)
+	}
+}
+
+// 1531: a 503 while another job is open is said as busy, with no loop and no retry when brother left no job.
+func TestABusyScannerIsSaidAndNotRetriedWhenBrotherLeftNoJob(t *testing.T) {
+	f := newFakeScanner(t, fixture(t, "scan.jpg"))
+	f.open = true // another client's job
+	c := NewClient(f.base())
+	c.RetryWait = time.Millisecond
+	_, err := c.Scan(context.Background(), 300, "color")
+	if err == nil || !strings.Contains(err.Error(), "the scanner is busy with another job") {
+		t.Fatalf("error = %v, want busy", err)
+	}
+	if f.posts != 1 {
+		t.Fatalf("%d posts, want 1", f.posts)
+	}
+}
+
+// 1531: a job brother left open (it could not close it) is closed after a wait, and the refused POST goes once more.
+func TestABusyScannerWithABrotherJobLeftIsRetriedOnceAfterItCloses(t *testing.T) {
+	f := newFakeScanner(t, fixture(t, "scan.jpg"))
+	f.stuck = true
+	c := NewClient(f.base())
+	c.RetryWait = time.Millisecond
+	if _, err := c.Scan(context.Background(), 300, "color"); err == nil {
+		t.Fatal("a stuck job is a scan")
+	}
+	f.mu.Lock()
+	if !f.open {
+		t.Fatal("the fake should have kept the job open")
+	}
+	f.stuck = false
+	f.mu.Unlock()
+	c2 := NewClient(f.base())
+	c2.RetryWait = time.Millisecond
+	if _, err := c2.Scan(context.Background(), 300, "color"); err != nil {
+		t.Fatalf("scan after the left job closes: %v", err)
+	}
+	if f.posts != 3 {
+		t.Fatalf("%d posts, want 3 (stuck job, refused, retried)", f.posts)
+	}
+}
+
+// 1531: a second page for a Platen job is an error, never a silently incomplete scan, and the job is closed.
+func TestMorePagesThanTheGlassHoldsIsAnErrorAndTheJobIsClosed(t *testing.T) {
+	f := newFakeScanner(t, fixture(t, "scan.jpg"))
+	f.pages = 2
+	image, err := NewClient(f.base()).Scan(context.Background(), 300, "color")
+	if err == nil || !strings.Contains(err.Error(), "the scanner gave more pages than the glass holds") {
+		t.Fatalf("image %d bytes, error = %v, want more pages than the glass holds", len(image), err)
+	}
+	if f.open || f.deletes != 1 {
+		t.Fatalf("job open %v, %d DELETEs, want closed by 1 DELETE", f.open, f.deletes)
+	}
+}

modules/brother/cmd/brother/helpers_test.go

@@ -29,6 +29,16 @@ type fakeScanner struct {
 	documents int // NextDocument requests
 	refuse    int // when set, the status ScanJobs answers instead of 201
 	noPage    bool
+	// A real scanner keeps a job open (and answers a new POST with 503) until its NextDocument has answered 404
+	// or the job is DELETEd; the fake does the same.
+	open      bool
+	delivered bool
+	pages     int // pages a job holds beyond the first
+	served    int
+	deletes   int
+	gone      bool // the first NextDocument answers 410 (the job was cancelled at the device)
+	dropConn  bool // the first NextDocument loses its answer
+	stuck     bool // the job cannot be closed: NextDocument and DELETE both fail
 }
 
 func newFakeScanner(t *testing.T, image []byte) *fakeScanner {
@@ -58,6 +68,11 @@ func newFakeScanner(t *testing.T, image []byte) *fakeScanner {
 			http.Error(w, "busy", f.refuse)
 			return
 		}
+		if f.open {
+			http.Error(w, "busy", http.StatusServiceUnavailable)
+			return
+		}
+		f.open, f.delivered, f.served = true, false, 0
 		// The captured Location names the real scanner's host; the test's own server stands in for it.
 		w.Header().Set("Location", f.URL+"/eSCL/ScanJobs/7320ead5")
 		w.WriteHeader(http.StatusCreated)
@@ -66,13 +81,47 @@ func newFakeScanner(t *testing.T, image []byte) *fakeScanner {
 		f.mu.Lock()
 		defer f.mu.Unlock()
 		f.documents++
-		if f.documents > 1 || f.noPage {
+		if f.stuck {
+			http.Error(w, "stuck", http.StatusInternalServerError)
+			return
+		}
+		if f.gone {
+			f.gone, f.open = false, false
+			http.Error(w, "gone", http.StatusGone)
+			return
+		}
+		if f.dropConn {
+			if hj, ok := w.(http.Hijacker); ok {
+				c, _, _ := hj.Hijack()
+				c.Close()
+			}
+			return
+		}
+		if f.noPage || (f.delivered && f.served > f.pages) {
+			f.open = false
 			http.NotFound(w, r)
 			return
 		}
+		f.delivered = true
+		f.served++
 		w.Header().Set("Content-Type", "image/jpeg")
 		w.Write(f.image)
 	})
+	mux.HandleFunc("/eSCL/ScanJobs/7320ead5", func(w http.ResponseWriter, r *http.Request) {
+		f.mu.Lock()
+		defer f.mu.Unlock()
+		if r.Method != http.MethodDelete {
+			http.Error(w, "delete only", http.StatusMethodNotAllowed)
+			return
+		}
+		f.deletes++
+		if f.stuck {
+			http.Error(w, "stuck", http.StatusInternalServerError)
+			return
+		}
+		f.open = false
+		w.WriteHeader(http.StatusOK)
+	})
 	f.Server = httptest.NewServer(mux)
 	t.Cleanup(f.Close)
 	return f

modules/brother/cmd/brother/service_test.go

@@ -140,7 +140,7 @@ func TestTheSameScanTwiceIsOneObject(t *testing.T) {
 		t.Fatal(err)
 	}
 	f.mu.Lock()
-	f.documents = 0
+	_ = f.documents
 	f.mu.Unlock()
 	if _, err := r.scan(t); err != nil {
 		t.Fatal(err)