diff --git a/modules/mesh-files/cmd/mesh-files/files.go b/modules/mesh-files/cmd/mesh-files/files.go index 98dffcd1..e001e3e0 100644 --- a/modules/mesh-files/cmd/mesh-files/files.go +++ b/modules/mesh-files/cmd/mesh-files/files.go @@ -74,6 +74,8 @@ type Files struct { // download use them (transfer.go). Tests replace both. Link func(bucket, key, method string, seconds int) (string, error) HTTP *http.Client + // Bound is how long a transfer may take; zero is TransferTime (a test makes it shorter). + Bound time.Duration mu sync.Mutex } diff --git a/modules/mesh-files/cmd/mesh-files/transfer.go b/modules/mesh-files/cmd/mesh-files/transfer.go index 4307e8cf..c91e1512 100644 --- a/modules/mesh-files/cmd/mesh-files/transfer.go +++ b/modules/mesh-files/cmd/mesh-files/transfer.go @@ -27,12 +27,30 @@ import ( "git.novox.be/novox/mesh-sdk/go/objectlink" ) -// LinkSeconds is how long the link asked for one transfer lives, and TransferTime how long a transfer may take. +// LinkSeconds is how long the link asked for one transfer lives, and TransferTime how long a transfer may take: the +// tool runner cuts every tool call at 30 s, so a transfer that is not done in 25 s is refused rather than left to run +// on after the caller was told it failed. Nothing is kept of a transfer that did not finish. const ( LinkSeconds = 300 - TransferTime = 5 * time.Minute + TransferTime = 25 * time.Second ) +func (f *Files) bound() time.Duration { + if f.Bound > 0 { + return f.Bound + } + return TransferTime +} + +// tooSlow says a transfer that ran out of time in words, naming the bound. +func (f *Files) tooSlow(err error, what string) error { + if errors.Is(err, context.DeadlineExceeded) || strings.Contains(err.Error(), "context deadline exceeded") || + strings.Contains(err.Error(), "Client.Timeout") { + return fmt.Errorf("%s did not finish within %s (a tool call is cut at 30 s), so it is refused and nothing is kept", what, f.bound()) + } + return err +} + var sha256Key = regexp.MustCompile(`^[0-9a-f]{64}$`) // askLink asks the object store's seat for one link (ADR 0468 ยง8). The answer is {link, method, seconds, expires-at}. @@ -68,7 +86,7 @@ func (f *Files) client() *http.Client { if f.HTTP != nil { return f.HTTP } - return &http.Client{Timeout: TransferTime} + return &http.Client{Timeout: f.bound()} } // UploadAnswer is what upload answers. @@ -82,6 +100,8 @@ type UploadAnswer struct { // Upload sends one file of this node into a bucket. The key is the file's SHA-256 when none is given; the hash is // then made first (one read of the file), so the link is asked for the key it will be used for. func (f *Files) Upload(raw, bucket, key string) (UploadAnswer, error) { + ctx, cancel := context.WithTimeout(context.Background(), f.bound()) + defer cancel() if strings.TrimSpace(bucket) == "" { return UploadAnswer{}, errors.New("bucket is needed: name the bucket the file goes into") } @@ -123,11 +143,9 @@ func (f *Files) Upload(raw, bucket, key string) (UploadAnswer, error) { if err != nil { return UploadAnswer{}, err } - ctx, cancel := context.WithTimeout(context.Background(), TransferTime) - defer cancel() res, err := objectlink.Put(ctx, f.client(), link, in, info.Size(), objectlink.MaxDocument) if err != nil { - return UploadAnswer{}, fmt.Errorf("%s could not be sent: %w", path, err) + return UploadAnswer{}, f.tooSlow(fmt.Errorf("%s could not be sent: %w", path, err), "the upload of "+path) } if want != "" && res.SHA256 != want { return UploadAnswer{}, fmt.Errorf("%s changed while it was sent: it was stored under %s but hashes to %s", path, want, res.SHA256) @@ -206,11 +224,11 @@ func (f *Files) Download(a Act, bucket, key, raw, wantSHA string) (DownloadAnswe os.Remove(tmp.Name()) } }() - ctx, cancel := context.WithTimeout(context.Background(), TransferTime) + ctx, cancel := context.WithTimeout(context.Background(), f.bound()) defer cancel() res, err := objectlink.Get(ctx, f.client(), link, tmp, objectlink.MaxDocument, expect) if err != nil { - return DownloadAnswer{}, fmt.Errorf("%s/%s could not be fetched: %w", bucket, key, err) + return DownloadAnswer{}, f.tooSlow(fmt.Errorf("%s/%s could not be fetched: %w", bucket, key, err), "the download of "+bucket+"/"+key) } if err := tmp.Close(); err != nil { return DownloadAnswer{}, err diff --git a/modules/mesh-files/cmd/mesh-files/transfer_test.go b/modules/mesh-files/cmd/mesh-files/transfer_test.go index aff55cad..ffc71864 100644 --- a/modules/mesh-files/cmd/mesh-files/transfer_test.go +++ b/modules/mesh-files/cmd/mesh-files/transfer_test.go @@ -12,6 +12,7 @@ import ( "strings" "sync" "testing" + "time" ) // fakeStore stands for the object store behind signed links: a PUT keeps the body under its path, a GET answers it. @@ -260,3 +261,41 @@ func TestTheSeatListsUploadAndDownload(t *testing.T) { t.Fatalf("tools %v", names) } } + +// 1597/AC9: a transfer slower than its bound is refused naming it, and no object or file is left. +func TestASlowTransferIsRefusedNamingTheBoundAndKeepsNothing(t *testing.T) { + g := testFiles(t) + g.Bound = 150 * time.Millisecond + release := make(chan struct{}) + slow := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet { + w.Write([]byte("half of the document")) + w.(http.Flusher).Flush() + } + select { // a store that does not finish: nothing is kept of a PUT whose body it never took whole + case <-release: + case <-r.Context().Done(): + } + })) + t.Cleanup(func() { close(release); slow.Close() }) + g.Link = func(bucket, key, method string, seconds int) (string, error) { + return slow.URL + "/" + bucket + "/" + key, nil + } + write(t, g.path("in.bin"), "a document") + if _, err := g.Upload(g.path("in.bin"), "b", "k"); err == nil || !strings.Contains(err.Error(), "did not finish within 150ms") || + !strings.Contains(err.Error(), "nothing is kept") { + t.Fatalf("upload error = %v", err) + } + if err := os.MkdirAll(g.path("out"), 0o755); err != nil { + t.Fatal(err) + } + if _, err := g.Download(by, "b", "k", g.path("out/doc"), ""); err == nil || !strings.Contains(err.Error(), "did not finish within 150ms") { + t.Fatalf("download error = %v", err) + } + if e, _ := os.ReadDir(g.path("out")); len(e) != 0 { + t.Errorf("a file is left: %v", e) + } + if TransferTime != 25*time.Second { + t.Errorf("the bound is %s, want 25s", TransferTime) + } +}