Loading cmd/felis/api.go +24 −0 Changes for cmd/felis/api.go: 24 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -494,6 +494,9 @@ func cmdAPI(args []string, stdout, stderr io.Writer) int { if fileStage != nil { go expireFileSessions(ctx, fileStage, fileSessionSweep, stderr) } if exporter != nil { go expireExports(ctx, a, exportSweep) } go retention.Loop(ctx, drv.DB(), retention.Policy{Audit: auditRetention}, retentionInterval, slog.Default()) servers := []*http.Server{internalSrv, externalSrv} Loading Loading @@ -824,6 +827,27 @@ func expireFileSessions(ctx context.Context, s *fileedit.Stage, every time.Durat } } // exportSweep is how often expireExports runs: an export whose Job never // connected is stopped within a minute of going stale. const exportSweep = time.Minute // expireExports runs the export sweep (api.API.ExpireExports) on a ticker. The // export routes sweep as they are called, and an owner who closed the tab calls // none; a Job whose Pod never got going would then keep the server from // starting until the Job's deadline. func expireExports(ctx context.Context, a interface{ ExpireExports() }, every time.Duration) { t := time.NewTicker(every) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: a.ExpireExports() } } } // reapRejectedContexts deletes, once an hour, the uploaded contexts of // submissions rejected more than submit.RejectedContextRetention ago, and the // chunked uploads left untouched for submit.StalePartRetention. Without it a Loading cmd/felis/api_test.go +25 −0 Changes for cmd/felis/api_test.go: 25 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -286,3 +286,28 @@ func TestExpireFileSessions(t *testing.T) { t.Fatalf("said %q", got) } } type sweepCount struct{ n atomic.Int32 } func (s *sweepCount) ExpireExports() { s.n.Add(1) } // TestExpireExports: the loop sweeps on each tick, and returns once felis-api // shuts down. func TestExpireExports(t *testing.T) { var s sweepCount ctx, cancel := context.WithCancel(context.Background()) done := make(chan struct{}) go func() { expireExports(ctx, &s, time.Millisecond); close(done) }() for deadline := time.Now().Add(5 * time.Second); s.n.Load() < 3; time.Sleep(time.Millisecond) { if time.Now().After(deadline) { cancel() t.Fatalf("swept %d times in 5s at a 1ms tick", s.n.Load()) } } cancel() select { case <-done: case <-time.After(5 * time.Second): t.Fatal("the loop outlived its context") } } cmd/felis/export.go +30 −8 Changes for cmd/felis/export.go: 30 added lines, 8 removed lines. Original line number Diff line number Diff line Loading @@ -3,6 +3,7 @@ package main import ( "context" "crypto/sha256" "encoding/base64" "encoding/hex" "encoding/json" "errors" Loading @@ -14,6 +15,7 @@ import ( "net/http" "os" "os/signal" "strconv" "strings" "syscall" "time" Loading Loading @@ -67,6 +69,7 @@ func cmdExport(args []string, stdout, stderr io.Writer) int { fmt.Fprintf(stderr, "felis export: --target-url and %s are required\n", worldexport.TokenEnv) return 2 } limitHeapToCgroup() ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() Loading Loading @@ -222,19 +225,30 @@ func (d *digestReader) Read(p []byte) (int, error) { return n, err } // streamExport runs write straight into the body of the PUT. An error from // write aborts the chunked body, and felis-api then cuts the browser's download // off rather than end it; that error is the one reported, since the PUT's own // error only wraps it. When the PUT ends first, write is stopped. // streamExport runs write straight into the body of the PUT, hashing it as it // goes. Once write has finished, the SHA-256 of all it wrote rides the // request's trailer (worldexport.DigestTrailer), and felis-api holds back the // last bytes from the browser until what it received hashes the same. An error // from write aborts the chunked body before the trailer, and felis-api then // cuts the browser's download off rather than end it; that error is the one // reported, since the PUT's own error only wraps it. When the PUT ends first, // write is stopped. func streamExport(ctx context.Context, target, token, contentType string, size int64, write func(io.Writer) error) error { pr, pw := io.Pipe() trailer := http.Header{worldexport.DigestTrailer: nil} werr := make(chan error, 1) go func() { err := write(pw) sum := sha256.New() err := write(io.MultiWriter(pw, sum)) if err == nil { // Set before the body ends: the transport reads the trailer once it // has read the body to its end. trailer.Set(worldexport.DigestTrailer, "sha-256=:"+base64.StdEncoding.EncodeToString(sum.Sum(nil))+":") } pw.CloseWithError(err) werr <- err }() err := putExport(ctx, target, token, contentType, pr, size) err := putExport(ctx, target, token, contentType, pr, size, trailer) pr.CloseWithError(io.ErrClosedPipe) if w := <-werr; w != nil && !errors.Is(w, io.ErrClosedPipe) { return w Loading @@ -248,12 +262,20 @@ func streamExport(ctx context.Context, target, token, contentType string, size i // redirects. felis-api answers only after the whole download, which the Job's // activeDeadlineSeconds bounds, so the header timeout is a backstop for a // wedged endpoint and not the real limit. func putExport(ctx context.Context, target, token, contentType string, body io.Reader, size int64) error { // // The body always goes chunked, which is what lets it end with a trailer; a // size the Job knows (-1 when it does not) goes as worldexport.LengthHeader in // place of Content-Length. func putExport(ctx context.Context, target, token, contentType string, body io.Reader, size int64, trailer http.Header) error { req, err := http.NewRequestWithContext(ctx, http.MethodPut, target, body) if err != nil { return err } req.ContentLength = size req.ContentLength = -1 req.Trailer = trailer if size >= 0 { req.Header.Set(worldexport.LengthHeader, strconv.FormatInt(size, 10)) } req.Header.Set("Authorization", "Bearer "+token) req.Header.Set("Content-Type", contentType) client := &http.Client{ Loading cmd/felis/export_test.go +25 −2 Changes for cmd/felis/export_test.go: 25 added lines, 2 removed lines. Original line number Diff line number Diff line Loading @@ -8,6 +8,7 @@ import ( "context" "crypto/rand" "crypto/sha256" "encoding/base64" "encoding/hex" "errors" "io" Loading Loading @@ -54,6 +55,25 @@ func receiveExport(t *testing.T, reply func(w http.ResponseWriter)) *exportRecei func noContent(w http.ResponseWriter) { w.WriteHeader(http.StatusNoContent) } // sentWhole fails unless the upload rcv got ended with the Content-Digest // trailer of its own bytes, and declared length as its size (-1: none). func sentWhole(t *testing.T, rcv *exportReceiver, length int64) { t.Helper() sum := sha256.Sum256(rcv.body) want := "sha-256=:" + base64.StdEncoding.EncodeToString(sum[:]) + ":" wantLength := "" if length >= 0 { wantLength = strconv.FormatInt(length, 10) } r := rcv.req if rcv.readErr != nil || r.Trailer.Get(worldexport.DigestTrailer) != want || r.Header.Get(worldexport.LengthHeader) != wantLength || r.ContentLength != -1 || strings.Join(r.TransferEncoding, ",") != "chunked" { t.Fatalf("upload read %v, trailer %v, %s %q, length %d, encoding %v; want trailer %q and %s %q, chunked", rcv.readErr, r.Trailer, worldexport.LengthHeader, r.Header.Get(worldexport.LengthHeader), r.ContentLength, r.TransferEncoding, want, worldexport.LengthHeader, wantLength) } } func tarEntries(t *testing.T, archive []byte) map[string]string { t.Helper() gz, err := gzip.NewReader(bytes.NewReader(archive)) Loading Loading @@ -141,6 +161,7 @@ func TestCmdExportWorld(t *testing.T) { if got := tarEntries(t, rcv.body); !reflect.DeepEqual(got, want) { t.Fatalf("archive holds %v\nwant %v", got, want) } sentWhole(t, rcv, -1) want2 := "felis export: left out 1 entries a tar cannot hold (symbolic links, devices, sockets)\n" + "felis export: left out 2 files that hold platform secrets\n" + "felis export: server=survival mode=world downloaded\n" Loading Loading @@ -297,6 +318,7 @@ func TestCmdExportBackup(t *testing.T) { if got := tarEntries(t, rcv.body); !reflect.DeepEqual(got, want) { t.Fatalf("archive holds %d entries, want exactly the redacted properties, level.dat and the region file", len(got)) } sentWhole(t, rcv, -1) if want := "felis export: left out 1 files that hold platform secrets\nfelis export: server=survival mode=backup downloaded\n"; stdout.String() != want { t.Errorf("stdout = %q, want %q", stdout.String(), want) } Loading Loading @@ -417,9 +439,10 @@ func TestCmdExportFiles(t *testing.T) { if code != 0 || stdout != "felis export: server=survival mode=files downloaded\n" { t.Fatalf("exit %d, stdout %q, stderr %q", code, stdout, stderr) } if string(rcv.body) != want || rcv.req.ContentLength != int64(len(want)) || rcv.req.Header.Get("Content-Type") != "application/octet-stream" { t.Fatalf("body %q, length %d, type %q; want %q", rcv.body, rcv.req.ContentLength, rcv.req.Header.Get("Content-Type"), want) if string(rcv.body) != want || rcv.req.Header.Get("Content-Type") != "application/octet-stream" { t.Fatalf("body %q, type %q; want %q", rcv.body, rcv.req.Header.Get("Content-Type"), want) } sentWhole(t, rcv, int64(len(want))) }) } Loading cmd/felis/files.go +35 −1 Changes for cmd/felis/files.go: 35 added lines, 1 removed line. Original line number Diff line number Diff line Loading @@ -57,6 +57,7 @@ func cmdFiles(args []string, stdout, stderr io.Writer) int { fmt.Fprintln(stderr, "felis files: --op is required") return 2 } limitHeapToCgroup() req := fileedit.Request{ Op: *op, Path: *path, To: *to, Expect: *expect, CreateOnly: *createOnly, Overwrite: *overwrite, } Loading Loading @@ -86,6 +87,13 @@ func cmdFiles(args []string, stdout, stderr io.Writer) int { req.Upload = &fileedit.Upload{ Size: *size, SHA256: *sum, Open: func() (io.ReadCloser, error) { return fetchUpload(ctx, *sourceURL, token) }, Landed: func() { if err := reportLanded(ctx, *sourceURL, token); err != nil { // The file is in place; felis-api drops its copy when it // has sat idle long enough, and the panel cancels it too. fmt.Fprintf(stderr, "felis files: tell felis-api the upload landed: %v\n", err) } }, } } // An upload or an unzip (the only ops that report progress) can run long Loading @@ -112,7 +120,7 @@ func cmdFiles(args []string, stdout, stderr io.Writer) int { // fetchUpload opens the staged upload on felis-api's internal face. There is no // retry: the token opens the upload once (fileedit.Stage), so a second attempt // could only be refused, and the caller retries the failed Job whole (a file // sent in parts stays staged until it has been served whole once, so that retry // sent in parts stays staged until its Job reports it landed, so that retry // does not send it again). Redirects are refused because the request carries the // token and the internal face never redirects; the header timeout catches a // wedged endpoint, and the Job's activeDeadlineSeconds bounds the body. Loading @@ -136,3 +144,29 @@ func fetchUpload(ctx context.Context, url, token string) (io.ReadCloser, error) } return resp.Body, nil } // reportLanded tells felis-api the upload's file is in place (DELETE on the URL // it was fetched from, with the same token), so it deletes the copy it staged. // One try: the file has landed whatever the answer, and a copy nobody deletes // is dropped once it has sat idle for fileedit.SessionIdle. func reportLanded(ctx context.Context, url, token string) error { ctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() req, err := http.NewRequestWithContext(ctx, http.MethodDelete, url, nil) if err != nil { return err } req.Header.Set("Authorization", "Bearer "+token) client := &http.Client{ CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, } resp, err := client.Do(req) if err != nil { return err } resp.Body.Close() if resp.StatusCode != http.StatusNoContent { return fmt.Errorf("DELETE returned %s", resp.Status) } return nil } Loading
cmd/felis/api.go +24 −0 Changes for cmd/felis/api.go: 24 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -494,6 +494,9 @@ func cmdAPI(args []string, stdout, stderr io.Writer) int { if fileStage != nil { go expireFileSessions(ctx, fileStage, fileSessionSweep, stderr) } if exporter != nil { go expireExports(ctx, a, exportSweep) } go retention.Loop(ctx, drv.DB(), retention.Policy{Audit: auditRetention}, retentionInterval, slog.Default()) servers := []*http.Server{internalSrv, externalSrv} Loading Loading @@ -824,6 +827,27 @@ func expireFileSessions(ctx context.Context, s *fileedit.Stage, every time.Durat } } // exportSweep is how often expireExports runs: an export whose Job never // connected is stopped within a minute of going stale. const exportSweep = time.Minute // expireExports runs the export sweep (api.API.ExpireExports) on a ticker. The // export routes sweep as they are called, and an owner who closed the tab calls // none; a Job whose Pod never got going would then keep the server from // starting until the Job's deadline. func expireExports(ctx context.Context, a interface{ ExpireExports() }, every time.Duration) { t := time.NewTicker(every) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: a.ExpireExports() } } } // reapRejectedContexts deletes, once an hour, the uploaded contexts of // submissions rejected more than submit.RejectedContextRetention ago, and the // chunked uploads left untouched for submit.StalePartRetention. Without it a Loading
cmd/felis/api_test.go +25 −0 Changes for cmd/felis/api_test.go: 25 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -286,3 +286,28 @@ func TestExpireFileSessions(t *testing.T) { t.Fatalf("said %q", got) } } type sweepCount struct{ n atomic.Int32 } func (s *sweepCount) ExpireExports() { s.n.Add(1) } // TestExpireExports: the loop sweeps on each tick, and returns once felis-api // shuts down. func TestExpireExports(t *testing.T) { var s sweepCount ctx, cancel := context.WithCancel(context.Background()) done := make(chan struct{}) go func() { expireExports(ctx, &s, time.Millisecond); close(done) }() for deadline := time.Now().Add(5 * time.Second); s.n.Load() < 3; time.Sleep(time.Millisecond) { if time.Now().After(deadline) { cancel() t.Fatalf("swept %d times in 5s at a 1ms tick", s.n.Load()) } } cancel() select { case <-done: case <-time.After(5 * time.Second): t.Fatal("the loop outlived its context") } }
cmd/felis/export.go +30 −8 Changes for cmd/felis/export.go: 30 added lines, 8 removed lines. Original line number Diff line number Diff line Loading @@ -3,6 +3,7 @@ package main import ( "context" "crypto/sha256" "encoding/base64" "encoding/hex" "encoding/json" "errors" Loading @@ -14,6 +15,7 @@ import ( "net/http" "os" "os/signal" "strconv" "strings" "syscall" "time" Loading Loading @@ -67,6 +69,7 @@ func cmdExport(args []string, stdout, stderr io.Writer) int { fmt.Fprintf(stderr, "felis export: --target-url and %s are required\n", worldexport.TokenEnv) return 2 } limitHeapToCgroup() ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() Loading Loading @@ -222,19 +225,30 @@ func (d *digestReader) Read(p []byte) (int, error) { return n, err } // streamExport runs write straight into the body of the PUT. An error from // write aborts the chunked body, and felis-api then cuts the browser's download // off rather than end it; that error is the one reported, since the PUT's own // error only wraps it. When the PUT ends first, write is stopped. // streamExport runs write straight into the body of the PUT, hashing it as it // goes. Once write has finished, the SHA-256 of all it wrote rides the // request's trailer (worldexport.DigestTrailer), and felis-api holds back the // last bytes from the browser until what it received hashes the same. An error // from write aborts the chunked body before the trailer, and felis-api then // cuts the browser's download off rather than end it; that error is the one // reported, since the PUT's own error only wraps it. When the PUT ends first, // write is stopped. func streamExport(ctx context.Context, target, token, contentType string, size int64, write func(io.Writer) error) error { pr, pw := io.Pipe() trailer := http.Header{worldexport.DigestTrailer: nil} werr := make(chan error, 1) go func() { err := write(pw) sum := sha256.New() err := write(io.MultiWriter(pw, sum)) if err == nil { // Set before the body ends: the transport reads the trailer once it // has read the body to its end. trailer.Set(worldexport.DigestTrailer, "sha-256=:"+base64.StdEncoding.EncodeToString(sum.Sum(nil))+":") } pw.CloseWithError(err) werr <- err }() err := putExport(ctx, target, token, contentType, pr, size) err := putExport(ctx, target, token, contentType, pr, size, trailer) pr.CloseWithError(io.ErrClosedPipe) if w := <-werr; w != nil && !errors.Is(w, io.ErrClosedPipe) { return w Loading @@ -248,12 +262,20 @@ func streamExport(ctx context.Context, target, token, contentType string, size i // redirects. felis-api answers only after the whole download, which the Job's // activeDeadlineSeconds bounds, so the header timeout is a backstop for a // wedged endpoint and not the real limit. func putExport(ctx context.Context, target, token, contentType string, body io.Reader, size int64) error { // // The body always goes chunked, which is what lets it end with a trailer; a // size the Job knows (-1 when it does not) goes as worldexport.LengthHeader in // place of Content-Length. func putExport(ctx context.Context, target, token, contentType string, body io.Reader, size int64, trailer http.Header) error { req, err := http.NewRequestWithContext(ctx, http.MethodPut, target, body) if err != nil { return err } req.ContentLength = size req.ContentLength = -1 req.Trailer = trailer if size >= 0 { req.Header.Set(worldexport.LengthHeader, strconv.FormatInt(size, 10)) } req.Header.Set("Authorization", "Bearer "+token) req.Header.Set("Content-Type", contentType) client := &http.Client{ Loading
cmd/felis/export_test.go +25 −2 Changes for cmd/felis/export_test.go: 25 added lines, 2 removed lines. Original line number Diff line number Diff line Loading @@ -8,6 +8,7 @@ import ( "context" "crypto/rand" "crypto/sha256" "encoding/base64" "encoding/hex" "errors" "io" Loading Loading @@ -54,6 +55,25 @@ func receiveExport(t *testing.T, reply func(w http.ResponseWriter)) *exportRecei func noContent(w http.ResponseWriter) { w.WriteHeader(http.StatusNoContent) } // sentWhole fails unless the upload rcv got ended with the Content-Digest // trailer of its own bytes, and declared length as its size (-1: none). func sentWhole(t *testing.T, rcv *exportReceiver, length int64) { t.Helper() sum := sha256.Sum256(rcv.body) want := "sha-256=:" + base64.StdEncoding.EncodeToString(sum[:]) + ":" wantLength := "" if length >= 0 { wantLength = strconv.FormatInt(length, 10) } r := rcv.req if rcv.readErr != nil || r.Trailer.Get(worldexport.DigestTrailer) != want || r.Header.Get(worldexport.LengthHeader) != wantLength || r.ContentLength != -1 || strings.Join(r.TransferEncoding, ",") != "chunked" { t.Fatalf("upload read %v, trailer %v, %s %q, length %d, encoding %v; want trailer %q and %s %q, chunked", rcv.readErr, r.Trailer, worldexport.LengthHeader, r.Header.Get(worldexport.LengthHeader), r.ContentLength, r.TransferEncoding, want, worldexport.LengthHeader, wantLength) } } func tarEntries(t *testing.T, archive []byte) map[string]string { t.Helper() gz, err := gzip.NewReader(bytes.NewReader(archive)) Loading Loading @@ -141,6 +161,7 @@ func TestCmdExportWorld(t *testing.T) { if got := tarEntries(t, rcv.body); !reflect.DeepEqual(got, want) { t.Fatalf("archive holds %v\nwant %v", got, want) } sentWhole(t, rcv, -1) want2 := "felis export: left out 1 entries a tar cannot hold (symbolic links, devices, sockets)\n" + "felis export: left out 2 files that hold platform secrets\n" + "felis export: server=survival mode=world downloaded\n" Loading Loading @@ -297,6 +318,7 @@ func TestCmdExportBackup(t *testing.T) { if got := tarEntries(t, rcv.body); !reflect.DeepEqual(got, want) { t.Fatalf("archive holds %d entries, want exactly the redacted properties, level.dat and the region file", len(got)) } sentWhole(t, rcv, -1) if want := "felis export: left out 1 files that hold platform secrets\nfelis export: server=survival mode=backup downloaded\n"; stdout.String() != want { t.Errorf("stdout = %q, want %q", stdout.String(), want) } Loading Loading @@ -417,9 +439,10 @@ func TestCmdExportFiles(t *testing.T) { if code != 0 || stdout != "felis export: server=survival mode=files downloaded\n" { t.Fatalf("exit %d, stdout %q, stderr %q", code, stdout, stderr) } if string(rcv.body) != want || rcv.req.ContentLength != int64(len(want)) || rcv.req.Header.Get("Content-Type") != "application/octet-stream" { t.Fatalf("body %q, length %d, type %q; want %q", rcv.body, rcv.req.ContentLength, rcv.req.Header.Get("Content-Type"), want) if string(rcv.body) != want || rcv.req.Header.Get("Content-Type") != "application/octet-stream" { t.Fatalf("body %q, type %q; want %q", rcv.body, rcv.req.Header.Get("Content-Type"), want) } sentWhole(t, rcv, int64(len(want))) }) } Loading
cmd/felis/files.go +35 −1 Changes for cmd/felis/files.go: 35 added lines, 1 removed line. Original line number Diff line number Diff line Loading @@ -57,6 +57,7 @@ func cmdFiles(args []string, stdout, stderr io.Writer) int { fmt.Fprintln(stderr, "felis files: --op is required") return 2 } limitHeapToCgroup() req := fileedit.Request{ Op: *op, Path: *path, To: *to, Expect: *expect, CreateOnly: *createOnly, Overwrite: *overwrite, } Loading Loading @@ -86,6 +87,13 @@ func cmdFiles(args []string, stdout, stderr io.Writer) int { req.Upload = &fileedit.Upload{ Size: *size, SHA256: *sum, Open: func() (io.ReadCloser, error) { return fetchUpload(ctx, *sourceURL, token) }, Landed: func() { if err := reportLanded(ctx, *sourceURL, token); err != nil { // The file is in place; felis-api drops its copy when it // has sat idle long enough, and the panel cancels it too. fmt.Fprintf(stderr, "felis files: tell felis-api the upload landed: %v\n", err) } }, } } // An upload or an unzip (the only ops that report progress) can run long Loading @@ -112,7 +120,7 @@ func cmdFiles(args []string, stdout, stderr io.Writer) int { // fetchUpload opens the staged upload on felis-api's internal face. There is no // retry: the token opens the upload once (fileedit.Stage), so a second attempt // could only be refused, and the caller retries the failed Job whole (a file // sent in parts stays staged until it has been served whole once, so that retry // sent in parts stays staged until its Job reports it landed, so that retry // does not send it again). Redirects are refused because the request carries the // token and the internal face never redirects; the header timeout catches a // wedged endpoint, and the Job's activeDeadlineSeconds bounds the body. Loading @@ -136,3 +144,29 @@ func fetchUpload(ctx context.Context, url, token string) (io.ReadCloser, error) } return resp.Body, nil } // reportLanded tells felis-api the upload's file is in place (DELETE on the URL // it was fetched from, with the same token), so it deletes the copy it staged. // One try: the file has landed whatever the answer, and a copy nobody deletes // is dropped once it has sat idle for fileedit.SessionIdle. func reportLanded(ctx context.Context, url, token string) error { ctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() req, err := http.NewRequestWithContext(ctx, http.MethodDelete, url, nil) if err != nil { return err } req.Header.Set("Authorization", "Bearer "+token) client := &http.Client{ CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, } resp, err := client.Do(req) if err != nil { return err } resp.Body.Close() if resp.StatusCode != http.StatusNoContent { return fmt.Errorf("DELETE returned %s", resp.Status) } return nil }