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
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()}
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 {
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,
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
}
// 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
// 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).
Upload func(node, path, bucket string) (string, error)
Bucket func() (string, error)
// Fetch reads the links the device process signs; DocExpiry and DocRetry are the open documents' waits. They are
// 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
@@ -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 {
// 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
}
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
gctx, cancel := context.WithTimeout(ctx, 5*time.Minute)
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)
}
pdf := buf.Bytes()
@@ -329,7 +334,28 @@ func withoutLink(ans map[string]any) map[string]any {
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{}
// 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") {
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 == "" {
continue
}
left := docExpiry
left := s.docExpiry()
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)
}