feature/taildrop: replace outgoing-file progress channel with synchronous reporter
serveFilePut tracked outgoing-file progress through an unbuffered progressUpdates channel whose close was owned by the request goroutine while writers were spread across manifest parsing, the progresstracking.Reader callback, singleFilePut failure paths, and the success path. That writer-closes mismatch made the send-on-closed-channel panic effectively unfixable in place. Replace it with a request-scoped outgoingProgress reporter. Transfer code reports state by method call; the reporter coalesces hot-path updates and is flushed once via defer in serveFilePut. With no producer channel to close, the panic is structurally impossible. Fixes #19115 Fixes #19817 Change-Id: I8f00d982d2c79880dfc1f8104c5eed06e94b5a6c Signed-off-by: James Tucker <james@tailscale.com>
This commit is contained in:
committed by
James Tucker
parent
f277bfb09d
commit
d1912167dc
@@ -9,7 +9,6 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"maps"
|
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"runtime"
|
"runtime"
|
||||||
"slices"
|
"slices"
|
||||||
@@ -411,14 +410,16 @@ func (e *Extension) taildropTargetStatus(p tailcfg.NodeView, nb ipnext.NodeBacke
|
|||||||
return ipnstate.TaildropTargetAvailable
|
return ipnstate.TaildropTargetAvailable
|
||||||
}
|
}
|
||||||
|
|
||||||
// updateOutgoingFiles updates b.outgoingFiles to reflect the given updates and
|
// updateOutgoingFiles merges updates into e.outgoingFiles and emits an
|
||||||
// sends an ipn.Notify with the full list of outgoingFiles.
|
// ipn.Notify.
|
||||||
func (e *Extension) updateOutgoingFiles(updates map[string]*ipn.OutgoingFile) {
|
func (e *Extension) updateOutgoingFiles(updates map[string]ipn.OutgoingFile) {
|
||||||
e.mu.Lock()
|
e.mu.Lock()
|
||||||
if e.outgoingFiles == nil {
|
if e.outgoingFiles == nil {
|
||||||
e.outgoingFiles = make(map[string]*ipn.OutgoingFile, len(updates))
|
e.outgoingFiles = make(map[string]*ipn.OutgoingFile, len(updates))
|
||||||
}
|
}
|
||||||
maps.Copy(e.outgoingFiles, updates)
|
for id, f := range updates {
|
||||||
|
e.outgoingFiles[id] = &f
|
||||||
|
}
|
||||||
outgoingFiles := make([]*ipn.OutgoingFile, 0, len(e.outgoingFiles))
|
outgoingFiles := make([]*ipn.OutgoingFile, 0, len(e.outgoingFiles))
|
||||||
for _, file := range e.outgoingFiles {
|
for _, file := range e.outgoingFiles {
|
||||||
outgoingFiles = append(outgoingFiles, file)
|
outgoingFiles = append(outgoingFiles, file)
|
||||||
|
|||||||
@@ -17,6 +17,7 @@ import (
|
|||||||
"net/url"
|
"net/url"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"tailscale.com/client/tailscale/apitype"
|
"tailscale.com/client/tailscale/apitype"
|
||||||
@@ -127,27 +128,9 @@ func serveFilePut(h *localapi.Handler, w http.ResponseWriter, r *http.Request) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Periodically report progress of outgoing files.
|
// Notify any updates buffered at request return.
|
||||||
outgoingFiles := make(map[string]*ipn.OutgoingFile)
|
progress := newOutgoingProgress(ext)
|
||||||
t := time.NewTicker(1 * time.Second)
|
defer progress.notify()
|
||||||
progressUpdates := make(chan ipn.OutgoingFile)
|
|
||||||
defer close(progressUpdates)
|
|
||||||
|
|
||||||
go func() {
|
|
||||||
defer t.Stop()
|
|
||||||
defer ext.updateOutgoingFiles(outgoingFiles)
|
|
||||||
for {
|
|
||||||
select {
|
|
||||||
case u, ok := <-progressUpdates:
|
|
||||||
if !ok {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
outgoingFiles[u.ID] = &u
|
|
||||||
case <-t.C:
|
|
||||||
ext.updateOutgoingFiles(outgoingFiles)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
switch r.Method {
|
switch r.Method {
|
||||||
case "PUT":
|
case "PUT":
|
||||||
@@ -157,16 +140,73 @@ func serveFilePut(h *localapi.Handler, w http.ResponseWriter, r *http.Request) {
|
|||||||
Name: filenameEscaped,
|
Name: filenameEscaped,
|
||||||
DeclaredSize: r.ContentLength,
|
DeclaredSize: r.ContentLength,
|
||||||
}
|
}
|
||||||
singleFilePut(h, r.Context(), progressUpdates, w, r.Body, dstURL, file)
|
singleFilePut(h, r.Context(), progress, w, r.Body, dstURL, file)
|
||||||
case "POST":
|
case "POST":
|
||||||
multiFilePost(h, progressUpdates, w, r, peerID, dstURL)
|
multiFilePost(h, progress, w, r, peerID, dstURL)
|
||||||
default:
|
default:
|
||||||
http.Error(w, "want PUT to put file", http.StatusBadRequest)
|
http.Error(w, "want PUT to put file", http.StatusBadRequest)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func multiFilePost(h *localapi.Handler, progressUpdates chan (ipn.OutgoingFile), w http.ResponseWriter, r *http.Request, peerID tailcfg.StableNodeID, dstURL *url.URL) {
|
// outgoingProgress forwards file-put progress to the Taildrop Extension
|
||||||
|
// for one localapi request. update coalesces hot-path changes; notify
|
||||||
|
// distributes any pending updates to observers immediately, disregarding
|
||||||
|
// the coalescing interval. The owner must call notify before returning
|
||||||
|
// so buffered updates aren't lost.
|
||||||
|
//
|
||||||
|
// outgoingProgress is safe for concurrent use.
|
||||||
|
type outgoingProgress struct {
|
||||||
|
ext *Extension
|
||||||
|
notifyInterval time.Duration
|
||||||
|
|
||||||
|
mu sync.Mutex
|
||||||
|
pending map[string]ipn.OutgoingFile // by OutgoingFile.ID
|
||||||
|
last time.Time
|
||||||
|
}
|
||||||
|
|
||||||
|
func newOutgoingProgress(ext *Extension) *outgoingProgress {
|
||||||
|
return &outgoingProgress{
|
||||||
|
ext: ext,
|
||||||
|
notifyInterval: time.Second,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// update buffers f. If notifyInterval has elapsed since the last notify,
|
||||||
|
// pending updates are also distributed to observers.
|
||||||
|
func (p *outgoingProgress) update(f ipn.OutgoingFile) {
|
||||||
|
var updates map[string]ipn.OutgoingFile
|
||||||
|
p.mu.Lock()
|
||||||
|
mak.Set(&p.pending, f.ID, f)
|
||||||
|
if time.Since(p.last) >= p.notifyInterval {
|
||||||
|
updates, p.pending = p.pending, nil
|
||||||
|
p.last = time.Now()
|
||||||
|
}
|
||||||
|
p.mu.Unlock()
|
||||||
|
if updates != nil {
|
||||||
|
p.ext.updateOutgoingFiles(updates)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// notify distributes any pending updates to observers immediately,
|
||||||
|
// disregarding the coalescing interval. Callers should notify
|
||||||
|
// explicitly for new files and completion events so observers don't
|
||||||
|
// have to wait for the next coalesced send. It is safe to call
|
||||||
|
// repeatedly.
|
||||||
|
func (p *outgoingProgress) notify() {
|
||||||
|
var updates map[string]ipn.OutgoingFile
|
||||||
|
p.mu.Lock()
|
||||||
|
if len(p.pending) > 0 {
|
||||||
|
updates, p.pending = p.pending, nil
|
||||||
|
p.last = time.Now()
|
||||||
|
}
|
||||||
|
p.mu.Unlock()
|
||||||
|
if updates != nil {
|
||||||
|
p.ext.updateOutgoingFiles(updates)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func multiFilePost(h *localapi.Handler, progress *outgoingProgress, w http.ResponseWriter, r *http.Request, peerID tailcfg.StableNodeID, dstURL *url.URL) {
|
||||||
_, params, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
|
_, params, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
http.Error(w, fmt.Sprintf("invalid Content-Type for multipart POST: %s", err), http.StatusBadRequest)
|
http.Error(w, fmt.Sprintf("invalid Content-Type for multipart POST: %s", err), http.StatusBadRequest)
|
||||||
@@ -209,13 +249,13 @@ func multiFilePost(h *localapi.Handler, progressUpdates chan (ipn.OutgoingFile),
|
|||||||
|
|
||||||
for _, file := range manifest {
|
for _, file := range manifest {
|
||||||
outgoingFilesByName[file.Name] = file
|
outgoingFilesByName[file.Name] = file
|
||||||
progressUpdates <- file
|
progress.update(file)
|
||||||
}
|
}
|
||||||
|
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
if !singleFilePut(h, r.Context(), progressUpdates, ww, part, dstURL, outgoingFilesByName[part.FileName()]) {
|
if !singleFilePut(h, r.Context(), progress, ww, part, dstURL, outgoingFilesByName[part.FileName()]) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -271,22 +311,25 @@ func (ww *multiFilePostResponseWriter) Flush(w http.ResponseWriter) error {
|
|||||||
func singleFilePut(
|
func singleFilePut(
|
||||||
h *localapi.Handler,
|
h *localapi.Handler,
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
progressUpdates chan (ipn.OutgoingFile),
|
progress *outgoingProgress,
|
||||||
w http.ResponseWriter,
|
w http.ResponseWriter,
|
||||||
body io.Reader,
|
body io.Reader,
|
||||||
dstURL *url.URL,
|
dstURL *url.URL,
|
||||||
outgoingFile ipn.OutgoingFile,
|
outgoingFile ipn.OutgoingFile,
|
||||||
) bool {
|
) bool {
|
||||||
outgoingFile.Started = time.Now()
|
outgoingFile.Started = time.Now()
|
||||||
body = progresstracking.NewReader(body, 1*time.Second, func(n int, err error) {
|
progress.update(outgoingFile)
|
||||||
|
progress.notify()
|
||||||
|
body = progresstracking.NewReader(body, time.Second, func(n int, err error) {
|
||||||
outgoingFile.Sent = int64(n)
|
outgoingFile.Sent = int64(n)
|
||||||
progressUpdates <- outgoingFile
|
progress.update(outgoingFile)
|
||||||
})
|
})
|
||||||
|
|
||||||
fail := func() {
|
fail := func() {
|
||||||
outgoingFile.Finished = true
|
outgoingFile.Finished = true
|
||||||
outgoingFile.Succeeded = false
|
outgoingFile.Succeeded = false
|
||||||
progressUpdates <- outgoingFile
|
progress.update(outgoingFile)
|
||||||
|
progress.notify()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Before we PUT a file we check to see if there are any existing partial file and if so,
|
// Before we PUT a file we check to see if there are any existing partial file and if so,
|
||||||
@@ -351,7 +394,8 @@ func singleFilePut(
|
|||||||
|
|
||||||
outgoingFile.Finished = true
|
outgoingFile.Finished = true
|
||||||
outgoingFile.Succeeded = true
|
outgoingFile.Succeeded = true
|
||||||
progressUpdates <- outgoingFile
|
progress.update(outgoingFile)
|
||||||
|
progress.notify()
|
||||||
|
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user