Give the Service its own fetch client and open-document waits instead of reading package variables from background goroutines, which a test or reload changed meanwhile: data races under -race (issue 1601)
mesh/repo-check head 6cf51173d586 merged into trunk b2f5ebb27dbf beside mesh-controller#424@550cf37d — pass: its merge-check.sh passed, each part in its to…
mesh/merge-gate head 6cf51173d586 merged into trunk b2f5ebb27dbf — pass: group feat/files-through-the-store composed: every machine composes with the chang…
mesh/delivery superseded: its walk was superseded and names no walk that took it over: a walk that took it over answers its merge, and it follows that wa…
mesh/delivery-group group feat/files-through-the-store stopped: superseded, not carried by a later delivery: novox/mesh-catalog@6cf51173d586, novox/mesh-contro…

This commit is contained in:
2026-10-21 18:03:58 +02:00
parent ad85a979e3
commit 6cf51173d5
3 changed files with 36 additions and 10 deletions
+1 -1
View File
@@ -264,7 +264,7 @@ func TestAnOpenDocumentIsFoundAgainAfterARestartAndItsPagesAreDeletedWhenClosed(
docExpiry = 200 * time.Millisecond docExpiry = 200 * time.Millisecond
t.Cleanup(func() { docExpiry = old }) t.Cleanup(func() { docExpiry = old })
s := &Service{Settings: Settings{Scanner: f.base(), Folder: folder, Resolution: 200, ColorMode: "color"}, Now: func() time.Time { return fixedNow }, 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()} Asks: askBus, AnswerWait: 5 * time.Second, ScanWait: 5 * time.Second, AnswerPoll: 5 * time.Millisecond, Jobs: NewJobs(), Fetch: fetchClient, DocExpiry: docExpiry}
if err := s.Recover(context.Background()); err != nil { if err := s.Recover(context.Background()); err != nil {
t.Fatal(err) t.Fatal(err)
} }
+1 -1
View File
@@ -112,7 +112,7 @@ func tools(settingsFile string, now func() time.Time) []stdio.Tool {
} }
return &Service{Settings: s, Node: os.Getenv("MESH_NODE"), Now: now, Asks: askBus, return &Service{Settings: s, Node: os.Getenv("MESH_NODE"), Now: now, Asks: askBus,
AnswerWait: answerWait, ScanWait: scanWait, AnswerPoll: answerPoll, AnswerWait: answerWait, ScanWait: scanWait, AnswerPoll: answerPoll,
QuickWait: quickWait, FeederWait: feederWait, Jobs: jobs, Upload: uploadVia, QuickWait: quickWait, FeederWait: feederWait, Jobs: jobs, Upload: uploadVia, Fetch: fetchClient, DocExpiry: docExpiry, DocRetry: docRetry,
Bucket: func() (string, error) { return ReadBucket(os.Getenv("BROTHER_BUCKET")) }}, nil Bucket: func() (string, error) { return ReadBucket(os.Getenv("BROTHER_BUCKET")) }}, nil
} }
// ask is a tool that reads its arguments, and answers what the service gives. // ask is a tool that reads its arguments, and answers what the service gives.
+34 -8
View File
@@ -34,8 +34,13 @@ type Service struct {
Jobs *Jobs Jobs *Jobs
// Upload asks a machine's node-files to send one of its files into the bucket and answers the key; Bucket is // Upload asks a machine's node-files to send one of its files into the bucket and answers the key; Bucket is
// brother's bucket. Both are the tool side's (printer.print with path and node). // brother's bucket. Both are the tool side's (printer.print with path and node).
Upload func(node, path, bucket string) (string, error) // Fetch reads the links the device process signs; DocExpiry and DocRetry are the open documents' waits. They are
Bucket func() (string, error) // copied into the Service when it is made, because background work reads them while a test (or a reload) changes the
// package's own, which would race.
Fetch *http.Client
DocExpiry, DocRetry time.Duration
Upload func(node, path, bucket string) (string, error)
Bucket func() (string, error)
} }
// Jobs are the scans this tool process has started and not yet been asked the outcome of. A scan runs in the // Jobs are the scans this tool process has started and not yet been asked the outcome of. A scan runs in the
@@ -236,7 +241,7 @@ func (s *Service) finishScan(ctx context.Context, id string, wait time.Duration,
} }
if doc, _ := ans["document"].(string); doc != "" && ans["key"] == nil { 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. // A page was added to an open document: nothing to file yet, and the document is closed if it is left open.
s.armDocument(doc, docExpiry, 0) s.armDocument(doc, s.docExpiry(), 0)
return ans, nil return ans, nil
} }
if ask.Document != "" && ask.Finish { if ask.Document != "" && ask.Finish {
@@ -296,7 +301,7 @@ func (s *Service) fileOne(ctx context.Context, ans map[string]any) (map[string]a
var buf bytes.Buffer var buf bytes.Buffer
gctx, cancel := context.WithTimeout(ctx, 5*time.Minute) gctx, cancel := context.WithTimeout(ctx, 5*time.Minute)
defer cancel() defer cancel()
if _, err := objectlink.Get(gctx, fetchClient, link, &buf, objectlink.MaxDocument, sum); err != nil { if _, err := objectlink.Get(gctx, s.fetchClient(), link, &buf, objectlink.MaxDocument, sum); err != nil {
return nil, fmt.Errorf("the scan is kept as %s, but it could not be read from the store: %w", key, err) return nil, fmt.Errorf("the scan is kept as %s, but it could not be read from the store: %w", key, err)
} }
pdf := buf.Bytes() pdf := buf.Bytes()
@@ -329,7 +334,28 @@ func withoutLink(ans map[string]any) map[string]any {
return out return out
} }
// fetchClient reads the links the device process signs; a test puts a client of its own in its place. func (s *Service) fetchClient() *http.Client {
if s.Fetch != nil {
return s.Fetch
}
return &http.Client{}
}
func (s *Service) docExpiry() time.Duration {
if s.DocExpiry > 0 {
return s.DocExpiry
}
return 30 * time.Minute
}
func (s *Service) docRetry() time.Duration {
if s.DocRetry > 0 {
return s.DocRetry
}
return 5 * time.Minute
}
// fetchClientDefault is what a Service made by the tools reads the links with; a test puts a client of its own in its place before it makes the tools.
var fetchClient = &http.Client{} var fetchClient = &http.Client{}
// ScanStatus says whether the scanner answers, and what it is doing. A scanner that does not answer is an // ScanStatus says whether the scanner answers, and what it is doing. A scanner that does not answer is an
@@ -370,7 +396,7 @@ func (s *Service) armDocument(doc string, d time.Duration, attempts int) {
if err == nil || attempts >= 10 || strings.Contains(err.Error(), "is closed already") || strings.Contains(err.Error(), "its id is wrong") { if err == nil || attempts >= 10 || strings.Contains(err.Error(), "is closed already") || strings.Contains(err.Error(), "its id is wrong") {
return return
} }
s.armDocument(doc, docRetry, attempts+1) s.armDocument(doc, s.docRetry(), attempts+1)
}) })
} }
@@ -388,9 +414,9 @@ func (s *Service) Recover(ctx context.Context) error {
if id == "" { if id == "" {
continue continue
} }
left := docExpiry left := s.docExpiry()
if at, err := time.Parse(time.RFC3339, fmt.Sprint(m["updated"])); err == nil { if at, err := time.Parse(time.RFC3339, fmt.Sprint(m["updated"])); err == nil {
left = max(docExpiry-s.Now().Sub(at), time.Second) left = max(s.docExpiry()-s.Now().Sub(at), time.Second)
} }
s.armDocument(id, left, 0) s.armDocument(id, left, 0)
} }