Skip to content

Commit a3f390c

Browse files
authored
Merge pull request #4 from GPULab-AI/fix/sync-deploy-hardening
fix(serverless): harden sync/deploy — content-hash diff, verify, safety
2 parents cf31863 + b584e5f commit a3f390c

16 files changed

Lines changed: 689 additions & 135 deletions

‎README.md‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -218,13 +218,20 @@ gpulab serverless list --volume my-volume
218218
### Syncing a directory to a volume
219219

220220
```bash
221-
# rsync-style: uploads only new/changed files, previews the diff, and warns how
222-
# many endpoints/replicas use the volume (they keep old code until restarted).
221+
# rsync-style: uploads only new/changed files (compared by content hash, so
222+
# equal-length edits are caught), hash-verifies each upload, previews the diff,
223+
# and warns how many endpoints/replicas use the volume (they keep old code until
224+
# restarted).
223225
gpulab volumes sync ./server my-volume server --dry-run # preview only
224226
gpulab volumes sync ./server my-volume server # upload changes
227+
gpulab volumes sync ./server my-volume server --size-only # faster, size-only compare
225228
gpulab volumes files upload my-volume ./local.py --to server/tasks.py # single file, renamed
226229
```
227230

231+
> `volumes sync --delete` refuses to run against an empty local dir (would wipe
232+
> the volume) unless you pass `--allow-empty`, and `--json`/non-interactive runs
233+
> won't mutate the volume without `--yes`.
234+
228235
## Templates (Docker images)
229236

230237
Create and manage reusable container templates. Only `--name` and `--image` are

‎internal/api/client.go‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package api
33
import (
44
"bytes"
55
"encoding/json"
6+
"errors"
67
"fmt"
78
"io"
89
"net/http"
@@ -157,10 +158,61 @@ func (c *Client) doRaw(method, path, contentType string, body io.Reader) ([]byte
157158
return respBody, nil
158159
}
159160

161+
// GetStream issues a GET and returns the raw response for streaming (the caller
162+
// must close the body). Non-2xx responses are turned into an *APIError with the
163+
// body read, so large downloads are never buffered on the success path.
164+
func (c *Client) GetStream(path string) (*http.Response, error) {
165+
req, err := http.NewRequest("GET", c.BaseURL+path, nil)
166+
if err != nil {
167+
return nil, fmt.Errorf("failed to create request: %w", err)
168+
}
169+
req.Header.Set("api-key", c.APIKey)
170+
req.Header.Set("Accept", "application/octet-stream")
171+
172+
if c.Debug {
173+
fmt.Fprintf(os.Stderr, "[DEBUG] GET %s (stream)\n", c.BaseURL+path)
174+
}
175+
176+
resp, err := c.HTTPClient.Do(req)
177+
if err != nil {
178+
return nil, fmt.Errorf("request failed: %w", err)
179+
}
180+
if resp.StatusCode >= 400 {
181+
body, _ := io.ReadAll(resp.Body)
182+
resp.Body.Close()
183+
var errResp struct {
184+
Message string `json:"message"`
185+
Error string `json:"error"`
186+
}
187+
json.Unmarshal(body, &errResp)
188+
message := errResp.Message
189+
if message == "" {
190+
message = errResp.Error
191+
}
192+
return nil, &APIError{StatusCode: resp.StatusCode, Message: message, Body: string(body)}
193+
}
194+
return resp, nil
195+
}
196+
197+
// PostStream posts a streaming body (e.g. a multipart upload built lazily) and
198+
// reads the (small) response fully.
199+
func (c *Client) PostStream(path, contentType string, body io.Reader) ([]byte, error) {
200+
return c.doRaw("POST", path, contentType, body)
201+
}
202+
160203
func (c *Client) Get(path string) ([]byte, error) {
161204
return c.do("GET", path, nil)
162205
}
163206

207+
// IsNotFound reports whether err is an API 404.
208+
func IsNotFound(err error) bool {
209+
var apiErr *APIError
210+
if errors.As(err, &apiErr) {
211+
return apiErr.StatusCode == 404
212+
}
213+
return false
214+
}
215+
164216
func (c *Client) Post(path string, body interface{}) ([]byte, error) {
165217
return c.do("POST", path, body)
166218
}

‎internal/api/volume_files.go‎

Lines changed: 61 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,8 @@
11
package api
22

33
import (
4-
"bytes"
4+
"crypto/sha256"
5+
"encoding/hex"
56
"encoding/json"
67
"fmt"
78
"io"
@@ -148,47 +149,74 @@ type UploadItem struct {
148149
// stored flat under destDir using its DestName (or basename). To place a file at
149150
// an arbitrary path, set destDir to the target directory and DestName to the
150151
// target filename.
152+
//
153+
// The multipart body is streamed through an io.Pipe so multi-GB files are never
154+
// buffered fully in memory.
151155
func (c *Client) UploadVolumeFiles(volumeUUID, destDir string, items []UploadItem) ([]byte, error) {
152-
var buf bytes.Buffer
153-
writer := multipart.NewWriter(&buf)
154-
155-
if err := writer.WriteField("path", destDir); err != nil {
156-
return nil, err
157-
}
158-
159-
for _, item := range items {
160-
file, err := os.Open(item.LocalPath)
156+
pr, pw := io.Pipe()
157+
writer := multipart.NewWriter(pw)
158+
159+
go func() {
160+
// Any error here is surfaced on the read side via CloseWithError, which
161+
// makes the HTTP request fail with the same error.
162+
err := func() error {
163+
if err := writer.WriteField("path", destDir); err != nil {
164+
return err
165+
}
166+
for _, item := range items {
167+
file, err := os.Open(item.LocalPath)
168+
if err != nil {
169+
return fmt.Errorf("failed to open %s: %w", item.LocalPath, err)
170+
}
171+
name := item.DestName
172+
if name == "" {
173+
name = filepath.Base(item.LocalPath)
174+
}
175+
part, err := writer.CreateFormFile("files[]", name)
176+
if err != nil {
177+
file.Close()
178+
return err
179+
}
180+
if _, err := io.Copy(part, file); err != nil {
181+
file.Close()
182+
return err
183+
}
184+
file.Close()
185+
// Flat upload: system-server stores each by its multipart filename.
186+
if err := writer.WriteField("relative_paths[]", ""); err != nil {
187+
return err
188+
}
189+
}
190+
return writer.Close()
191+
}()
161192
if err != nil {
162-
return nil, fmt.Errorf("failed to open %s: %w", item.LocalPath, err)
193+
pw.CloseWithError(err)
194+
return
163195
}
196+
pw.Close()
197+
}()
164198

165-
name := item.DestName
166-
if name == "" {
167-
name = filepath.Base(item.LocalPath)
168-
}
169-
part, err := writer.CreateFormFile("files[]", name)
170-
if err != nil {
171-
file.Close()
172-
return nil, err
173-
}
174-
if _, err := io.Copy(part, file); err != nil {
175-
file.Close()
176-
return nil, err
177-
}
178-
file.Close()
199+
return c.PostStream(c.volumeFilesPath(volumeUUID, "/upload"), writer.FormDataContentType(), pr)
200+
}
179201

180-
// Flat upload: keep files at the destination root (system-server stores
181-
// each by its multipart filename).
182-
if err := writer.WriteField("relative_paths[]", ""); err != nil {
183-
return nil, err
202+
// HashVolumeFile streams a file from the volume and returns its sha256 and size
203+
// without buffering it in memory. found=false (nil error) for a missing file.
204+
func (c *Client) HashVolumeFile(volumeUUID, path string) (sum string, size int64, found bool, err error) {
205+
resp, err := c.GetStream(c.volumeFilesPath(volumeUUID, "/download") + "?path=" + url.QueryEscape(path))
206+
if err != nil {
207+
if IsNotFound(err) {
208+
return "", 0, false, nil
184209
}
210+
return "", 0, false, err
185211
}
212+
defer resp.Body.Close()
186213

187-
if err := writer.Close(); err != nil {
188-
return nil, err
214+
h := sha256.New()
215+
n, err := io.Copy(h, resp.Body)
216+
if err != nil {
217+
return "", 0, false, err
189218
}
190-
191-
return c.doRaw("POST", c.volumeFilesPath(volumeUUID, "/upload"), writer.FormDataContentType(), &buf)
219+
return hex.EncodeToString(h.Sum(nil)), n, true, nil
192220
}
193221

194222
// UploadLocalFiles is a convenience wrapper for uploading files by path using
Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
package commands
2+
3+
import (
4+
"testing"
5+
6+
"github.com/GPULab-AI/gpulab-cli/internal/api"
7+
)
8+
9+
func TestDetectLogIssues_LineCountingAndAttributeError(t *testing.T) {
10+
logs := "" +
11+
"AttributeError: 'NoneType' object has no attribute 'size'\n" +
12+
"AttributeError: 'NoneType' object has no attribute 'size'\n" +
13+
"all good\n"
14+
issues := detectLogIssues(logs)
15+
16+
byCat := map[string]int{}
17+
for _, iss := range issues {
18+
byCat[iss.Category] = iss.Count
19+
}
20+
// The exact incident crash must be caught.
21+
if byCat["attribute-error"] != 2 {
22+
t.Fatalf("attribute-error count = %d, want 2", byCat["attribute-error"])
23+
}
24+
// Two overlapping patterns on the same line (NoneType twice) must count the
25+
// line once per category, not ×N.
26+
if byCat["nonetype"] != 2 {
27+
t.Fatalf("nonetype count = %d, want 2 (one per matching line)", byCat["nonetype"])
28+
}
29+
}
30+
31+
func TestComputeRequestStats_ErrorRateNeverExceeds100(t *testing.T) {
32+
code := func(c int) *int { return &c }
33+
msg := "boom"
34+
reqs := []api.ServerlessRequestLog{
35+
// A 5xx that is also an error status + error message — must count once.
36+
{Status: "error", ResponseStatusCode: code(500), ErrorMessage: msg},
37+
{Status: "completed", ResponseStatusCode: code(200)},
38+
}
39+
s := computeRequestStats(reqs)
40+
if s.Errors != 1 {
41+
t.Fatalf("deduped Errors = %d, want 1", s.Errors)
42+
}
43+
if s.ErrorRate != 50 {
44+
t.Fatalf("ErrorRate = %v, want 50", s.ErrorRate)
45+
}
46+
if s.ErrorRate > 100 {
47+
t.Fatalf("ErrorRate exceeds 100: %v", s.ErrorRate)
48+
}
49+
}

‎internal/commands/filesync.go‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
package commands
2+
3+
import (
4+
"crypto/sha256"
5+
"encoding/hex"
6+
"fmt"
7+
"io"
8+
"os"
9+
10+
"github.com/GPULab-AI/gpulab-cli/internal/api"
11+
)
12+
13+
// localFileDigest streams a local file and returns its sha256 hex digest and size
14+
// without buffering it in memory.
15+
func localFileDigest(path string) (string, int64, error) {
16+
f, err := os.Open(path)
17+
if err != nil {
18+
return "", 0, err
19+
}
20+
defer f.Close()
21+
h := sha256.New()
22+
n, err := io.Copy(h, f)
23+
if err != nil {
24+
return "", 0, err
25+
}
26+
return hex.EncodeToString(h.Sum(nil)), n, nil
27+
}
28+
29+
// remoteDiffersFromLocal decides whether localPath needs to be uploaded to
30+
// remotePath given the already-known remote size. It short-circuits on size
31+
// (different size ⇒ changed) and only content-hashes when the sizes match — the
32+
// exact case a size-only diff gets wrong (an equal-length edit). With sizeOnly
33+
// it trusts size alone.
34+
//
35+
// remoteExists=false means the file isn't on the volume yet (always "new").
36+
func remoteDiffersFromLocal(client *api.Client, uuid, remotePath, localPath string, localSize, remoteSize int64, remoteExists, sizeOnly bool) (changed bool, reason string, err error) {
37+
if !remoteExists {
38+
return true, "new", nil
39+
}
40+
if remoteSize != localSize {
41+
return true, "size", nil
42+
}
43+
if sizeOnly {
44+
return false, "", nil
45+
}
46+
// Same size — the dangerous case. Content-hash both sides.
47+
localSum, _, err := localFileDigest(localPath)
48+
if err != nil {
49+
return false, "", err
50+
}
51+
remoteSum, _, found, err := client.HashVolumeFile(uuid, remotePath)
52+
if err != nil {
53+
return false, "", fmt.Errorf("hashing remote %s: %w", remotePath, err)
54+
}
55+
if !found {
56+
return true, "new", nil
57+
}
58+
if remoteSum != localSum {
59+
return true, "content", nil
60+
}
61+
return false, "", nil
62+
}
63+
64+
// verifyReadback confirms the file uploaded to remotePath is byte-identical to
65+
// localPath by comparing sha256 digests, streaming both sides (no full-file
66+
// buffering, so it is safe for multi-GB files).
67+
func verifyReadback(client *api.Client, uuid, remotePath, localPath string) error {
68+
localSum, _, err := localFileDigest(localPath)
69+
if err != nil {
70+
return err
71+
}
72+
remoteSum, _, found, err := client.HashVolumeFile(uuid, remotePath)
73+
if err != nil {
74+
return fmt.Errorf("verify: cannot read back %s: %w", remotePath, err)
75+
}
76+
if !found {
77+
return fmt.Errorf("verify: %s is missing after upload", remotePath)
78+
}
79+
if remoteSum != localSum {
80+
return fmt.Errorf("verify FAILED for %s: readback differs from local (upload corrupted)", remotePath)
81+
}
82+
return nil
83+
}

‎internal/commands/logdetect.go‎

Lines changed: 18 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,9 @@ var logIssuePatterns = []logIssuePattern{
2020
{"OutOfMemoryError", "cuda-oom"},
2121
{"CUDA error: out of memory", "cuda-oom"},
2222
{"torch.cuda.OutOfMemoryError", "cuda-oom"},
23+
{"device-side assert", "cuda-assert"},
24+
{"CUDA error", "cuda-error"},
25+
{"illegal memory access", "cuda-error"},
2326
{"OOMKilled", "oom-killed"},
2427
{"Segmentation fault", "segfault"},
2528
{"core dumped", "segfault"},
@@ -28,8 +31,12 @@ var logIssuePatterns = []logIssuePattern{
2831
{"Address already in use", "port-conflict"},
2932
{"Connection refused", "connection-refused"},
3033
{"RuntimeError", "runtime-error"},
31-
{"AssertionError", "assertion-error"},
34+
{"AttributeError", "attribute-error"},
35+
{"ValueError", "value-error"},
36+
{"TypeError", "type-error"},
37+
{"IndexError", "index-error"},
3238
{"KeyError", "key-error"},
39+
{"AssertionError", "assertion-error"},
3340
}
3441

3542
// logIssue summarizes one detected problem category in a log blob.
@@ -49,12 +56,18 @@ func detectLogIssues(logs string) []logIssue {
4956
counts := map[string]int{}
5057
samples := map[string]string{}
5158
for _, line := range strings.Split(logs, "\n") {
59+
// Count each matching LINE once per category — overlapping patterns in
60+
// the same category (e.g. NoneType) must not inflate the count ×N.
61+
seen := map[string]bool{}
5262
for _, p := range logIssuePatterns {
5363
if strings.Contains(line, p.needle) {
54-
counts[p.category]++
55-
if samples[p.category] == "" {
56-
samples[p.category] = strings.TrimSpace(line)
57-
}
64+
seen[p.category] = true
65+
}
66+
}
67+
for cat := range seen {
68+
counts[cat]++
69+
if samples[cat] == "" {
70+
samples[cat] = strings.TrimSpace(line)
5871
}
5972
}
6073
}

0 commit comments

Comments
 (0)