Drop an ask taken up after its deadline, follow a scan for its run, re-arm open documents and delete their pages (issues 1539, 1552, 1538)

modules/brother/cmd/brother/deviceproc.go

@@ -102,6 +102,15 @@ func (w *deviceWorker) Handle(body []byte, early func()) (action, time.Duration)
 		w.logf("[brother] device: ask %s is %s old; nobody waits for it, the device is not touched", ask.ID, age.Round(time.Second))
 		return actionAck, 0
 	}
+	if !ask.Deadline.IsZero() && w.Now().After(ask.Deadline) {
+		// The asker has stopped waiting for this ask to be taken up: nothing is scanned or printed for it (issue
+		// 1539). The reason is its answer, for a late reader, and the message ends here.
+		w.logf("[brother] device: ask %s was taken up after its deadline %s; the device is not touched", ask.ID, ask.Deadline.Format(time.RFC3339))
+		if _, err := w.Answers.Create(ask.ID, map[string]any{"error": "the ask was taken up after its deadline (the device process was busy): nothing was done for it; ask again"}); err != nil && !errors.Is(err, stdio.ErrChanged) {
+			w.logf("[brother] device: the answer to %s could not be written: %v", ask.ID, err)
+		}
+		return actionTerm, 0
+	}
 	settled := false
 	if (ask.Op == OpScan || ask.Op == OpPrint) && early != nil {
 		early()
@@ -161,6 +170,8 @@ func (w *deviceWorker) run(ctx context.Context, ask Ask) (map[string]any, error)
 		return w.upload(ctx, ask)
 	case OpPrint:
 		return w.print(ctx, ask)
+	case OpOpen:
+		return w.openDocuments(ctx)
 	}
 	return nil, fmt.Errorf("the ask %s names an operation, %q, the device process does not do", ask.ID, ask.Op)
 }
@@ -573,6 +584,7 @@ type docPage struct {
 
 type docManifest struct {
 	Started  time.Time `json:"started"`
+	Updated  time.Time `json:"updated"` // the last page's time
 	Pages    []docPage `json:"pages"`
 	Finished string    `json:"finished,omitempty"` // the closed document's object key
 }
@@ -626,7 +638,7 @@ func (w *deviceWorker) scanDocument(ctx context.Context, ask Ask, svc *Service,
 	defer w.docs.Unlock()
 	var m *docManifest
 	if ask.Document == "start" {
-		m = &docManifest{Started: w.Now().UTC()}
+		m = &docManifest{Started: w.Now().UTC(), Updated: w.Now().UTC()}
 	} else {
 		var err error
 		if m, err = loadManifest(ctx, objects, id); err != nil {
@@ -647,6 +659,7 @@ func (w *deviceWorker) scanDocument(ctx context.Context, ask Ask, svc *Service,
 	if err := objects.Put(ctx, key, "image/jpeg", page.JPEG); err != nil {
 		return nil, err
 	}
+	m.Updated = w.Now().UTC()
 	m.Pages = append(m.Pages, docPage{Key: key, DPI: page.DPI, Width: page.Width, Height: page.Height, Cropped: page.Cropped})
 	if err := saveManifest(ctx, objects, id, m); err != nil {
 		return nil, err
@@ -685,5 +698,43 @@ func (w *deviceWorker) finishDocument(ctx context.Context, id string, m *docMani
 	if err := saveManifest(ctx, objects, id, m); err != nil {
 		return nil, fmt.Errorf("document %s is kept as %s, but could not be marked closed: %w", id, m.Finished, err)
 	}
+	// The pages are not needed any more: delete them. A bucket that refuses leaves them, and the answer says so.
+	for _, p := range m.Pages {
+		if err := objects.Delete(ctx, p.Key); err != nil {
+			answer["pending-kept"] = fmt.Sprintf("the pages of document %s could not be deleted from the bucket and stay under pending/%s/: %v", id, id, err)
+			break
+		}
+	}
 	return answer, nil
 }
+
+// openDocuments lists the documents of several pages that are open in the bucket, with the time of each one's last
+// page, so that the tool can arm their expiry again after it was restarted.
+func (w *deviceWorker) openDocuments(ctx context.Context) (map[string]any, error) {
+	objects, err := w.Objects()
+	if err != nil {
+		return nil, err
+	}
+	keys, err := objects.List(ctx, "pending/")
+	if err != nil {
+		return nil, err
+	}
+	open := []map[string]any{}
+	for _, k := range keys {
+		rest, ok := strings.CutPrefix(k, "pending/")
+		id, name, _ := strings.Cut(rest, "/")
+		if !ok || name != "manifest.json" || !uploadPattern.MatchString(id) {
+			continue
+		}
+		m, err := loadManifest(ctx, objects, id)
+		if err != nil {
+			continue // closed, or unreadable: not open
+		}
+		at := m.Updated
+		if at.IsZero() {
+			at = m.Started
+		}
+		open = append(open, map[string]any{"id": id, "pages": len(m.Pages), "updated": at.UTC().Format(time.RFC3339)})
+	}
+	return map[string]any{"open": open}, nil
+}

modules/brother/cmd/brother/glass_test.go

@@ -2,6 +2,9 @@ package main
 
 import (
 	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
 	"image"
 	"os"
 	"path/filepath"
@@ -182,3 +185,117 @@ func TestAFinishedGlassDocumentIsNotFiledAgainByItsExpiry(t *testing.T) {
 		t.Errorf("%d files, want 1", len(e))
 	}
 }
+
+// 1539: an ask carries its deadline, and a device process that takes it up later does nothing for it: nothing is
+// scanned, the answer says why, and the message ends.
+func TestAnAskTakenUpAfterItsDeadlineIsDroppedBeforeAnyWork(t *testing.T) {
+	f := newFakeScanner(t, fixture(t, "scan.jpg"))
+	r, folder := scanRig(t, f, Settings{})
+	scans := 0
+	r.worker.Capture = func(ctx context.Context, req ScanRequest) ([]Capture, error) {
+		scans++
+		return nil, errors.New("scanned")
+	}
+	late, _ := json.Marshal(Ask{ID: "late", Op: OpScan, At: fixedNow.Add(-4 * time.Minute), Deadline: fixedNow.Add(-time.Second)})
+	if act, _ := r.worker.Handle(late, func() { t.Error("a late ask was acknowledged early as work to be done") }); act != actionTerm {
+		t.Errorf("action = %v, want the message ended", act)
+	}
+	if scans != 0 || len(*r.said) != 0 {
+		t.Errorf("%d scans, %d events for a late ask", scans, len(*r.said))
+	}
+	if raw := r.answers.get("late"); !strings.Contains(string(raw), "deadline") {
+		t.Errorf("answer = %s", raw)
+	}
+	onTime, _ := json.Marshal(Ask{ID: "ontime", Op: OpScan, At: fixedNow, Deadline: fixedNow.Add(time.Minute)})
+	r.worker.Handle(onTime, nil)
+	if scans != 1 {
+		t.Errorf("an ask in time was not scanned")
+	}
+	_ = folder
+}
+
+// 1539: a scan the device process took up in time but ran past the tool's start-by wait is still filed: the tool
+// follows it for the run's own bound.
+func TestAScanThatRunsPastTheStartByWaitIsStillFiled(t *testing.T) {
+	f := newFakeScanner(t, fixture(t, "scan.jpg"))
+	r, folder := scanRig(t, f, Settings{})
+	old := scanWait
+	scanWait = 100 * time.Millisecond
+	t.Cleanup(func() { scanWait = old })
+	release := make(chan struct{})
+	r.worker.Capture = standInCapture(release, nil)
+	got, err := r.call(t, "scanner.scan", map[string]any{})
+	if err != nil || got.(map[string]any)["started"] != true {
+		t.Fatalf("answer = %v, %v, want started", got, err)
+	}
+	time.Sleep(300 * time.Millisecond) // past the start-by wait; the answer comes only now
+	close(release)
+	waitFor(t, "the late scan to be filed", func() bool { e, _ := os.ReadDir(folder); return len(e) == 1 })
+}
+
+// 1538: an expired document whose closing fails is not forgotten: it is tried again.
+func TestAnExpiredDocumentWhoseClosingFailsIsClosedOnTheNextTry(t *testing.T) {
+	f := newFakeScanner(t, fixture(t, "scan.jpg"))
+	r, folder := scanRig(t, f, Settings{Crop: false, Resolution: 200})
+	oldE, oldR := docExpiry, docRetry
+	docExpiry, docRetry = 100*time.Millisecond, 100*time.Millisecond
+	t.Cleanup(func() { docExpiry, docRetry = oldE, oldR })
+	askBus = &flakyBus{fakeAskBus: &fakeAskBus{worker: r.worker, answers: r.answers}, fails: 2,
+		match: func(a Ask) bool { return a.Op == OpScan && a.Finish }}
+	if _, err := r.call(t, "scanner.scan", docArgs("start", false)); err != nil {
+		t.Fatal(err)
+	}
+	waitFor(t, "the document to be filed on the third try", func() bool { e, _ := os.ReadDir(folder); return len(e) == 1 })
+}
+
+// 1538: after a restart of the tool process the open documents are found again in the bucket and their expiry
+// armed from the time of their last page; closing a document deletes its pages.
+func TestAnOpenDocumentIsFoundAgainAfterARestartAndItsPagesAreDeletedWhenClosed(t *testing.T) {
+	f := newFakeScanner(t, fixture(t, "scan.jpg"))
+	r, folder := scanRig(t, f, Settings{Crop: false, Resolution: 200})
+	got, err := r.call(t, "scanner.scan", docArgs("start", false))
+	if err != nil {
+		t.Fatal(err)
+	}
+	id := got.(map[string]any)["document"].(string)
+	// A new tool process: no timers. Its Recover finds the document and arms it.
+	old := docExpiry
+	docExpiry = 200 * time.Millisecond
+	t.Cleanup(func() { docExpiry = old })
+	s := &Service{Settings: Settings{Scanner: f.base(), Folder: folder, Resolution: 200, ColorMode: "color"}, Now: func() time.Time { return fixedNow },
+		Asks: askBus, AnswerWait: 5 * time.Second, ScanWait: 5 * time.Second, AnswerPoll: 5 * time.Millisecond, Jobs: NewJobs()}
+	if err := s.Recover(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	waitFor(t, "the recovered document to be closed", func() bool { e, _ := os.ReadDir(folder); return len(e) == 1 })
+	waitFor(t, "its pages to be deleted", func() bool {
+		for _, k := range r.objects.keys() {
+			if strings.HasPrefix(k, "pending/"+id+"/") && strings.HasSuffix(k, ".jpg") {
+				return false
+			}
+		}
+		return true
+	})
+	var hasManifest bool
+	for _, k := range r.objects.keys() {
+		hasManifest = hasManifest || k == "pending/"+id+"/manifest.json"
+	}
+	if !hasManifest {
+		t.Error("the manifest that says the document is closed was deleted too")
+	}
+}
+
+// 1552: a finished job's outcome is forgotten after jobKeep.
+func TestAFinishedJobIsForgottenAfterAnHour(t *testing.T) {
+	old := jobKeep
+	jobKeep = 30 * time.Millisecond
+	t.Cleanup(func() { jobKeep = old })
+	j := NewJobs()
+	first := j.put("a")
+	close(first.done)
+	time.Sleep(60 * time.Millisecond)
+	j.put("b")
+	if j.get("a") != nil || j.get("b") == nil {
+		t.Errorf("a = %v, b = %v", j.get("a"), j.get("b"))
+	}
+}

modules/brother/cmd/brother/main.go

@@ -322,6 +322,17 @@ func tools(settingsFile string, now func() time.Time) []stdio.Tool {
 				if err != nil {
 					return nil, err
 				}
+				// The node-engine asks this every minute, so it is also what brings back, after a restart of this
+				// process, the expiry of the documents left open in the bucket (issue 1538).
+				if s.Settings.Missing() == nil && jobs.tryRecover() {
+					go func() {
+						ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+						defer cancel()
+						if err := s.Recover(ctx); err != nil {
+							jobs.recoverFailed()
+						}
+					}()
+				}
 				if err := s.Settings.MissingPrinter(); err != nil {
 					return nil, err
 				}

modules/brother/cmd/brother/objects.go

@@ -14,6 +14,7 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"encoding/json"
+	"encoding/xml"
 	"errors"
 	"fmt"
 	"io"
@@ -35,6 +36,10 @@ type Objects interface {
 	Put(ctx context.Context, key, mediaType string, data []byte) error
 	// Get is the bytes kept under the key.
 	Get(ctx context.Context, key string) ([]byte, error)
+	// List is the keys kept that start with the prefix, in order (the first thousand).
+	List(ctx context.Context, prefix string) ([]string, error)
+	// Delete removes the key; a key not kept is not an error.
+	Delete(ctx context.Context, key string) error
 }
 
 // Store keeps the bytes under the key once: a key already kept is left as it is, since a key that names the
@@ -166,6 +171,37 @@ func (s *S3) Get(ctx context.Context, key string) ([]byte, error) {
 	return io.ReadAll(resp.Body)
 }
 
+func (s *S3) List(ctx context.Context, prefix string) ([]string, error) {
+	resp, err := s.do(ctx, http.MethodGet, "?list-type=2&prefix="+url.QueryEscape(prefix), "", nil)
+	if err != nil {
+		return nil, fmt.Errorf("the module's bucket cannot be reached: %w", err)
+	}
+	defer resp.Body.Close()
+	body, _ := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("the module's bucket answered %s to listing %s", resp.Status, prefix)
+	}
+	var out struct {
+		Keys []string `xml:"Contents>Key"`
+	}
+	if err := xml.Unmarshal(body, &out); err != nil {
+		return nil, fmt.Errorf("the module's bucket listed %s in a form that cannot be read: %w", prefix, err)
+	}
+	return out.Keys, nil
+}
+
+func (s *S3) Delete(ctx context.Context, key string) error {
+	resp, err := s.do(ctx, http.MethodDelete, key, "", nil)
+	if err != nil {
+		return fmt.Errorf("the module's bucket cannot be reached: %w", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode/100 != 2 && resp.StatusCode != http.StatusNotFound {
+		return fmt.Errorf("the module's bucket refused to delete %s: %s", key, resp.Status)
+	}
+	return nil
+}
+
 // Sign signs a request with AWS Signature Version 4 for S3: every header it carries but the agent and the length is
 // signed, with the payload's hash.
 func Sign(req *http.Request, body []byte, access, secret, region string, now time.Time) {

modules/brother/cmd/brother/objects_test.go

@@ -8,6 +8,7 @@ import (
 	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"sort"
 	"strings"
 	"sync"
 	"testing"
@@ -49,6 +50,24 @@ func (m *memObjects) Get(_ context.Context, key string) ([]byte, error) {
 	return b, nil
 }
 
+func (m *memObjects) List(_ context.Context, prefix string) ([]string, error) {
+	var out []string
+	for _, k := range m.keys() {
+		if strings.HasPrefix(k, prefix) {
+			out = append(out, k)
+		}
+	}
+	sort.Strings(out)
+	return out, nil
+}
+
+func (m *memObjects) Delete(_ context.Context, key string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	delete(m.kept, key)
+	return nil
+}
+
 func (m *memObjects) keys() []string {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -132,3 +151,28 @@ func TestTheStoresFactsAreRead(t *testing.T) {
 		t.Errorf("error = %v", err)
 	}
 }
+
+// 1538: the bucket lists keys by prefix (ListObjectsV2, signed, path-style) and deletes one.
+func TestTheBucketListsByPrefixAndDeletes(t *testing.T) {
+	var gotQuery, gotDelete string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.Method {
+		case http.MethodGet:
+			gotQuery = r.URL.RawQuery
+			w.Write([]byte(`<?xml version="1.0"?><ListBucketResult><Contents><Key>pending/a/manifest.json</Key></Contents><Contents><Key>pending/b/manifest.json</Key></Contents></ListBucketResult>`))
+		case http.MethodDelete:
+			gotDelete = r.URL.Path
+			w.WriteHeader(http.StatusNoContent)
+		}
+	}))
+	defer srv.Close()
+	at, port, _ := strings.Cut(strings.TrimPrefix(srv.URL, "http://"), ":")
+	s3 := NewS3(StoreFacts{Scheme: "http", At: at, Port: port, Bucket: "scans", Region: "us-east-1", Key: "AKID", Secret: "sek"})
+	keys, err := s3.List(context.Background(), "pending/")
+	if err != nil || len(keys) != 2 || keys[1] != "pending/b/manifest.json" || !strings.Contains(gotQuery, "prefix=pending%2F") || !strings.Contains(gotQuery, "list-type=2") {
+		t.Errorf("List = %v, %v (query %q)", keys, err, gotQuery)
+	}
+	if err := s3.Delete(context.Background(), "pending/a/001.jpg"); err != nil || gotDelete != "/scans/pending/a/001.jpg" {
+		t.Errorf("Delete: %v, %q", err, gotDelete)
+	}
+}

modules/brother/cmd/brother/printer.go

@@ -58,6 +58,8 @@ const (
 	OpDocument = "document"
 	OpUpload   = "upload"
 	OpPrint    = "print"
+	// OpOpen asks which documents of several pages are open in the bucket (issue 1538).
+	OpOpen = "open-documents"
 )
 
 // Ask is what a verb queues: the operation, when it was asked, and the operation's own arguments.
@@ -65,6 +67,9 @@ type Ask struct {
 	ID string    `json:"id"`
 	Op string    `json:"op"`
 	At time.Time `json:"asked-at"`
+	// Deadline is when the asker stops waiting for the device process to take the ask up: one taken after it is
+	// dropped before any work, so nothing is scanned or printed for an asker that has gone (issue 1539).
+	Deadline time.Time `json:"deadline,omitempty"`
 
 	Page   string `json:"page,omitempty"`   // settings
 	Links  bool   `json:"links,omitempty"`  // settings: the page's links are answered too
@@ -111,6 +116,10 @@ var (
 	scanWait   = ScanWait
 	quickWait  = QuickWait
 	feederWait = FeederWait
+	// scanRun and feederRun are how long the device process may take over a scan once it has taken it up (its own
+	// bounds, deviceproc.go); the tool follows a scan for its start-by wait and this.
+	scanRun    = 5 * time.Minute
+	feederRun  = FeederMaxBound + 5*time.Minute
 	answerPoll = 500 * time.Millisecond
 )
 
@@ -126,16 +135,20 @@ 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)
+	id, err := s.queue(ask, wait)
 	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) {
+// 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 startBy > 0 {
+		ask.Deadline = ask.At.Add(startBy)
+	}
 	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)
 	}

modules/brother/cmd/brother/scan.go

@@ -35,9 +35,10 @@ type Service struct {
 // 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
-	open map[string]*time.Timer // open documents: the timer that closes each
+	mu        sync.Mutex
+	m         map[string]*job
+	open      map[string]*time.Timer // open documents: the timer that closes each
+	recovered bool                   // the open documents of the bucket were looked for
 }
 
 // arm (re)starts the timer that closes the open document after d.
@@ -67,11 +68,18 @@ func (j *Jobs) disarm(doc string) {
 }
 
 type job struct {
+	at   time.Time
 	done chan struct{}
 	body map[string]any // set when done
 	err  error          // set when failed
 }
 
+// jobKeep is how long a finished job's outcome can be read with its id.
+var jobKeep = time.Hour
+
+// docRetry is how long after a failed closing of an expired document it is tried again.
+var docRetry = 5 * time.Minute
+
 // NewJobs is an empty set of jobs.
 func NewJobs() *Jobs { return &Jobs{m: map[string]*job{}} }
 
@@ -90,7 +98,17 @@ func (j *Jobs) put(id string) *job {
 	if j.m == nil {
 		j.m = map[string]*job{}
 	}
-	x := &job{done: make(chan struct{})}
+	// A job nobody has asked after for jobKeep is forgotten (a finished one only).
+	for k, o := range j.m {
+		select {
+		case <-o.done:
+			if time.Since(o.at) > jobKeep {
+				delete(j.m, k)
+			}
+		default:
+		}
+	}
+	x := &job{at: time.Now(), done: make(chan struct{})}
 	j.m[id] = x
 	return x
 }
@@ -111,17 +129,17 @@ func (s *Service) ScanNow(ctx context.Context, source string, duplex bool) (map[
 	if err != nil {
 		return nil, err
 	}
-	bound, quick := s.ScanWait, s.ScanWait
+	run, quick := scanRun, s.ScanWait
 	if req.Feeder {
-		bound, quick = s.FeederWait, s.QuickWait
+		run, quick = feederRun, s.QuickWait
 	}
-	return s.runScan(ctx, Ask{Op: OpScan, Source: source, Duplex: duplex}, bound, quick)
+	return s.runScan(ctx, Ask{Op: OpScan, Source: source, Duplex: duplex}, run, quick)
 }
 
 // runScan queues the ask and follows it in the background to its end (filing included); the caller is given the
 // outcome if it comes within quick, and started: true with the id otherwise.
-func (s *Service) runScan(ctx context.Context, ask Ask, bound, quick time.Duration) (map[string]any, error) {
-	id, j, err := s.startScan(ask, bound)
+func (s *Service) runScan(ctx context.Context, ask Ask, run, quick time.Duration) (map[string]any, error) {
+	id, j, err := s.startScan(ask, run)
 	if err != nil {
 		return nil, err
 	}
@@ -137,8 +155,14 @@ func (s *Service) runScan(ctx context.Context, ask Ask, bound, quick time.Durati
 }
 
 // startScan queues the ask and starts the goroutine that follows it, however long the caller waits.
-func (s *Service) startScan(ask Ask, bound time.Duration) (string, *job, error) {
-	id, err := s.queue(ask)
+//
+// The ask carries a deadline, ScanWait from now: the device process drops it unscanned if it is taken up later (it
+// may be busy with a long feeder scan). One taken up in time may run for run, and the goroutine waits for that
+// too, so a scan that is announced is always filed, however long the caller has waited.
+func (s *Service) startScan(ask Ask, run time.Duration) (string, *job, error) {
+	startBy := s.ScanWait
+	bound := startBy + run
+	id, err := s.queue(ask, startBy)
 	if err != nil {
 		return "", nil, err
 	}
@@ -171,7 +195,7 @@ func (s *Service) ScanPage(ctx context.Context, document string, finish bool) (m
 	if finish && document == "start" {
 		return nil, fmt.Errorf("finish closes an open document: name its id in document, not start")
 	}
-	return s.runScan(ctx, Ask{Op: OpScan, Document: document, Finish: finish}, s.ScanWait, s.ScanWait)
+	return s.runScan(ctx, Ask{Op: OpScan, Document: document, Finish: finish}, scanRun, s.ScanWait)
 }
 
 // ScanResult is the outcome of the scan with this id: its body and path when it is done, state running when it is
@@ -205,9 +229,7 @@ func (s *Service) finishScan(ctx context.Context, id string, wait time.Duration,
 	}
 	if doc, _ := ans["document"].(string); doc != "" && ans["key"] == nil {
 		// A page was added to an open document: nothing to file yet, and the document is closed if it is left open.
-		s.Jobs.arm(doc, docExpiry, func() {
-			_, _, _ = s.startScan(Ask{Op: OpScan, Document: doc, Finish: true}, s.ScanWait)
-		})
+		s.armDocument(doc, docExpiry, 0)
 		return ans, nil
 	}
 	if ask.Document != "" && ask.Finish {
@@ -334,3 +356,61 @@ func (s *Service) ScanList(limit int) (map[string]any, error) {
 	}
 	return map[string]any{"folder": s.Settings.Folder, "scans": scans}, nil
 }
+
+// armDocument closes the open document in d, from the expiry on: if that fails (the device process dropped the ask,
+// the bucket was away) it is tried again after docRetry, up to ten times, unless the document is gone or closed
+// already. A document is never forgotten while it can still be filed.
+func (s *Service) armDocument(doc string, d time.Duration, attempts int) {
+	s.Jobs.arm(doc, d, func() {
+		_, j, err := s.startScan(Ask{Op: OpScan, Document: doc, Finish: true}, scanRun)
+		if err == nil {
+			<-j.done
+			err = j.err
+		}
+		if err == nil || attempts >= 10 || strings.Contains(err.Error(), "is closed already") || strings.Contains(err.Error(), "its id is wrong") {
+			return
+		}
+		s.armDocument(doc, docRetry, attempts+1)
+	})
+}
+
+// Recover arms the expiry of every document the device process finds open in the bucket, so that a restart of
+// this process forgets none (issue 1538). A document gets what is left of its time since its last page.
+func (s *Service) Recover(ctx context.Context) error {
+	ans, err := s.call(ctx, Ask{Op: OpOpen}, s.AnswerWait)
+	if err != nil {
+		return err
+	}
+	open, _ := ans["open"].([]any)
+	for _, o := range open {
+		m, _ := o.(map[string]any)
+		id, _ := m["id"].(string)
+		if id == "" {
+			continue
+		}
+		left := docExpiry
+		if at, err := time.Parse(time.RFC3339, fmt.Sprint(m["updated"])); err == nil {
+			left = max(docExpiry-s.Now().Sub(at), time.Second)
+		}
+		s.armDocument(id, left, 0)
+	}
+	return nil
+}
+
+// tryRecover says whether this process has still to recover its open documents, and takes the turn; failed gives it
+// back.
+func (j *Jobs) tryRecover() bool {
+	j.mu.Lock()
+	defer j.mu.Unlock()
+	if j.recovered {
+		return false
+	}
+	j.recovered = true
+	return true
+}
+
+func (j *Jobs) recoverFailed() {
+	j.mu.Lock()
+	j.recovered = false
+	j.mu.Unlock()
+}

modules/brother/cmd/brother/service_test.go

@@ -342,6 +342,8 @@ type flakyBus struct {
 	*fakeAskBus
 	mu    sync.Mutex
 	fails int
+	// match says which asks fail; nil is the document fetches.
+	match func(Ask) bool
 }
 
 func (b *flakyBus) Publish(id string, body any) error {
@@ -349,7 +351,7 @@ func (b *flakyBus) Publish(id string, body any) error {
 	var ask Ask
 	json.Unmarshal(raw, &ask)
 	b.mu.Lock()
-	fail := ask.Op == OpDocument && b.fails > 0
+	fail := (b.match == nil && ask.Op == OpDocument || b.match != nil && b.match(ask)) && b.fails > 0
 	if fail {
 		b.fails--
 	}