Bound an upload and a download to 25 s, so a tool call is never cut while the transfer goes on (issue 1597)
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user