ipn{,/ipnlocal}, client/tailscale: move Taildrop recv notifications to LocalAPI HTTP method

Updates #6417

Change-Id: Iec544c477a0e5e9f1c6bf23555afec06255e2e22
Signed-off-by: Brad Fitzpatrick <bradfitz@tailscale.com>
pull/6322/head
Brad Fitzpatrick 2 years ago committed by Brad Fitzpatrick
parent f053f16460
commit 0f7da5c7dc

@ -426,8 +426,20 @@ func (lc *LocalClient) IDToken(ctx context.Context, aud string) (*tailcfg.TokenR
return decodeJSON[*tailcfg.TokenResponse](body) return decodeJSON[*tailcfg.TokenResponse](body)
} }
// WaitingFiles returns the list of received Taildrop files that have been
// received by the Tailscale daemon in its staging/cache directory but not yet
// transferred by the user's CLI or GUI client and written to a user's home
// directory somewhere.
func (lc *LocalClient) WaitingFiles(ctx context.Context) ([]apitype.WaitingFile, error) { func (lc *LocalClient) WaitingFiles(ctx context.Context) ([]apitype.WaitingFile, error) {
body, err := lc.get200(ctx, "/localapi/v0/files/") return lc.AwaitWaitingFiles(ctx, 0)
}
// AwaitWaitingFiles is like WaitingFiles but takes a duration to await for an answer.
// If the duration is 0, it will return immediately. The duration is respected at second
// granularity only. If no files are available, it returns (nil, nil).
func (lc *LocalClient) AwaitWaitingFiles(ctx context.Context, d time.Duration) ([]apitype.WaitingFile, error) {
path := "/localapi/v0/files/?waitsec=" + fmt.Sprint(int(d.Seconds()))
body, err := lc.get200(ctx, path)
if err != nil { if err != nil {
return nil, err return nil, err
} }

@ -26,7 +26,6 @@ import (
"golang.org/x/time/rate" "golang.org/x/time/rate"
"tailscale.com/client/tailscale/apitype" "tailscale.com/client/tailscale/apitype"
"tailscale.com/envknob" "tailscale.com/envknob"
"tailscale.com/ipn"
"tailscale.com/net/tsaddr" "tailscale.com/net/tsaddr"
"tailscale.com/tailcfg" "tailscale.com/tailcfg"
"tailscale.com/util/quarantine" "tailscale.com/util/quarantine"
@ -529,30 +528,16 @@ func wipeInbox(ctx context.Context) error {
} }
func waitForFile(ctx context.Context) error { func waitForFile(ctx context.Context) error {
c, bc, pumpCtx, cancel := connect(ctx) for {
defer cancel() ff, err := localClient.AwaitWaitingFiles(ctx, time.Hour)
fileWaiting := make(chan bool, 1) if len(ff) > 0 {
notifyError := make(chan error, 1) return nil
bc.SetNotifyCallback(func(n ipn.Notify) {
if n.ErrMessage != nil {
notifyError <- fmt.Errorf("Notify.ErrMessage: %v", *n.ErrMessage)
} }
if n.FilesWaiting != nil { if err := ctx.Err(); err != nil {
select { return err
case fileWaiting <- true: }
default: if err != nil && !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, context.Canceled) {
} return err
} }
})
go pump(pumpCtx, bc, c)
select {
case <-fileWaiting:
return nil
case <-pumpCtx.Done():
return pumpCtx.Err()
case <-ctx.Done():
return ctx.Err()
case err := <-notifyError:
return err
} }
} }

@ -76,6 +76,8 @@ type Notify struct {
// FilesWaiting if non-nil means that files are buffered in // FilesWaiting if non-nil means that files are buffered in
// the Tailscale daemon and ready for local transfer to the // the Tailscale daemon and ready for local transfer to the
// user's preferred storage location. // user's preferred storage location.
//
// Deprecated: use LocalClient.AwaitWaitingFiles instead.
FilesWaiting *empty.Message `json:",omitempty"` FilesWaiting *empty.Message `json:",omitempty"`
// IncomingFiles, if non-nil, specifies which files are in the // IncomingFiles, if non-nil, specifies which files are in the
@ -83,6 +85,8 @@ type Notify struct {
// Notify should not update the state of file transfers. A non-nil // Notify should not update the state of file transfers. A non-nil
// but empty IncomingFiles means that no files are in the middle // but empty IncomingFiles means that no files are in the middle
// of being transferred. // of being transferred.
//
// Deprecated: use LocalClient.AwaitWaitingFiles instead.
IncomingFiles []PartialFile `json:",omitempty"` IncomingFiles []PartialFile `json:",omitempty"`
// LocalTCPPort, if non-nil, informs the UI frontend which // LocalTCPPort, if non-nil, informs the UI frontend which

@ -180,7 +180,8 @@ type LocalBackend struct {
peerAPIListeners []*peerAPIListener peerAPIListeners []*peerAPIListener
loginFlags controlclient.LoginFlags loginFlags controlclient.LoginFlags
incomingFiles map[*incomingFile]bool incomingFiles map[*incomingFile]bool
lastStatusTime time.Time // status.AsOf value of the last processed status update fileWaiters map[*mapSetHandle]context.CancelFunc // handle => func to call on file received
lastStatusTime time.Time // status.AsOf value of the last processed status update
// directFileRoot, if non-empty, means to write received files // directFileRoot, if non-empty, means to write received files
// directly to this directory, without staging them in an // directly to this directory, without staging them in an
// intermediate buffered directory for "pick-up" later. If // intermediate buffered directory for "pick-up" later. If
@ -1709,6 +1710,9 @@ func (b *LocalBackend) sendFileNotify() {
var n ipn.Notify var n ipn.Notify
b.mu.Lock() b.mu.Lock()
for _, wakeWaiter := range b.fileWaiters {
wakeWaiter()
}
notifyFunc := b.notify notifyFunc := b.notify
apiSrv := b.peerAPIServer apiSrv := b.peerAPIServer
if notifyFunc == nil || apiSrv == nil { if notifyFunc == nil || apiSrv == nil {
@ -3579,6 +3583,20 @@ func (b *LocalBackend) TestOnlyPublicKeys() (machineKey key.MachinePublic, nodeK
return mk, nk return mk, nk
} }
// mapSetHandle is a minimal (but non-zero) value whose address serves as a map
// key for sets of non-comparable values that can't be map keys themselves.
type mapSetHandle byte
func (b *LocalBackend) setFileWaiter(handle *mapSetHandle, wakeWaiter context.CancelFunc) {
b.mu.Lock()
defer b.mu.Unlock()
if wakeWaiter == nil {
delete(b.fileWaiters, handle)
} else {
mak.Set(&b.fileWaiters, handle, wakeWaiter)
}
}
func (b *LocalBackend) WaitingFiles() ([]apitype.WaitingFile, error) { func (b *LocalBackend) WaitingFiles() ([]apitype.WaitingFile, error) {
b.mu.Lock() b.mu.Lock()
apiSrv := b.peerAPIServer apiSrv := b.peerAPIServer
@ -3586,6 +3604,42 @@ func (b *LocalBackend) WaitingFiles() ([]apitype.WaitingFile, error) {
return apiSrv.WaitingFiles() return apiSrv.WaitingFiles()
} }
// AwaitWaitingFiles is like WaitingFiles but blocks while ctx is not done,
// waiting for any files to be available.
//
// On return, exactly one of the results will be non-empty or non-nil,
// respectively.
func (b *LocalBackend) AwaitWaitingFiles(ctx context.Context) ([]apitype.WaitingFile, error) {
if ff, err := b.WaitingFiles(); err != nil || len(ff) > 0 {
return ff, err
}
for {
gotFile, gotFileCancel := context.WithCancel(context.Background())
defer gotFileCancel()
handle := new(mapSetHandle)
b.setFileWaiter(handle, gotFileCancel)
defer b.setFileWaiter(handle, nil)
// Now that we've registered ourselves, check again, in case
// of race. Otherwise there's a small window where we could
// miss a file arrival and wait forever.
if ff, err := b.WaitingFiles(); err != nil || len(ff) > 0 {
return ff, err
}
select {
case <-gotFile.Done():
if ff, err := b.WaitingFiles(); err != nil || len(ff) > 0 {
return ff, err
}
case <-ctx.Done():
return nil, ctx.Err()
}
}
}
func (b *LocalBackend) DeleteFile(name string) error { func (b *LocalBackend) DeleteFile(name string) error {
b.mu.Lock() b.mu.Lock()
apiSrv := b.peerAPIServer apiSrv := b.peerAPIServer

@ -7,6 +7,7 @@ package localapi
import ( import (
"bytes" "bytes"
"context"
"crypto/rand" "crypto/rand"
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
@ -680,8 +681,20 @@ func (h *Handler) serveFiles(w http.ResponseWriter, r *http.Request) {
http.Error(w, "want GET to list files", 400) http.Error(w, "want GET to list files", 400)
return return
} }
wfs, err := h.b.WaitingFiles() ctx := r.Context()
if err != nil { if s := r.FormValue("waitsec"); s != "" && s != "0" {
d, err := strconv.Atoi(s)
if err != nil {
http.Error(w, "invalid waitsec", http.StatusBadRequest)
return
}
deadline := time.Now().Add(time.Duration(d) * time.Second)
var cancel context.CancelFunc
ctx, cancel = context.WithDeadline(ctx, deadline)
defer cancel()
}
wfs, err := h.b.AwaitWaitingFiles(ctx)
if err != nil && ctx.Err() == nil {
http.Error(w, err.Error(), 500) http.Error(w, err.Error(), 500)
return return
} }

Loading…
Cancel
Save