diff --git a/modules/brother/cmd/brother/deviceproc.go b/modules/brother/cmd/brother/deviceproc.go index 335f9c1f..c0059f9e 100644 --- a/modules/brother/cmd/brother/deviceproc.go +++ b/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 +} diff --git a/modules/brother/cmd/brother/glass_test.go b/modules/brother/cmd/brother/glass_test.go index 974a3158..a5a13d69 100644 --- a/modules/brother/cmd/brother/glass_test.go +++ b/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")) + } +} diff --git a/modules/brother/cmd/brother/main.go b/modules/brother/cmd/brother/main.go index e6aff4bb..209008b6 100644 --- a/modules/brother/cmd/brother/main.go +++ 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 } diff --git a/modules/brother/cmd/brother/objects.go b/modules/brother/cmd/brother/objects.go index 6370f055..ffad97dc 100644 --- a/modules/brother/cmd/brother/objects.go +++ b/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) { diff --git a/modules/brother/cmd/brother/objects_test.go b/modules/brother/cmd/brother/objects_test.go index 2ba39fbc..1bcfbdc1 100644 --- a/modules/brother/cmd/brother/objects_test.go +++ b/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(`pending/a/manifest.jsonpending/b/manifest.json`)) + 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) + } +} diff --git a/modules/brother/cmd/brother/printer.go b/modules/brother/cmd/brother/printer.go index 071ac8b7..816a944e 100644 --- a/modules/brother/cmd/brother/printer.go +++ b/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) } diff --git a/modules/brother/cmd/brother/scan.go b/modules/brother/cmd/brother/scan.go index 8cabb9a7..fb57458a 100644 --- a/modules/brother/cmd/brother/scan.go +++ b/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() +} diff --git a/modules/brother/cmd/brother/service_test.go b/modules/brother/cmd/brother/service_test.go index f2de7080..11eea84c 100644 --- a/modules/brother/cmd/brother/service_test.go +++ b/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-- }