diff options
Diffstat (limited to 'internal/server/handlers/readcommand.go')
| -rw-r--r-- | internal/server/handlers/readcommand.go | 513 |
1 files changed, 462 insertions, 51 deletions
diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go index 44ba9e4..59254e8 100644 --- a/internal/server/handlers/readcommand.go +++ b/internal/server/handlers/readcommand.go @@ -1,58 +1,87 @@ package handlers import ( + "bytes" "context" + "io" "os" "path/filepath" "strings" "sync" "time" + "github.com/mimecast/dtail/internal/config" + "github.com/mimecast/dtail/internal/ctxutil" "github.com/mimecast/dtail/internal/io/dlog" "github.com/mimecast/dtail/internal/io/fs" - "github.com/mimecast/dtail/internal/io/line" + "github.com/mimecast/dtail/internal/io/journal" "github.com/mimecast/dtail/internal/lcontext" + "github.com/mimecast/dtail/internal/mapr/server" "github.com/mimecast/dtail/internal/omode" "github.com/mimecast/dtail/internal/regex" ) type readCommand struct { - server *ServerHandler - mode omode.Mode + server readCommandServer + mode omode.Mode + generation uint64 + shutdownCoordinator *shutdownCoordinator } -func newReadCommand(server *ServerHandler, mode omode.Mode) *readCommand { +type readStrategy func(context.Context, lcontext.LContext, fs.FileReader, regex.Regex) error + +type readProcessor interface { + ProcessLine(*bytes.Buffer, uint64, string) error + Flush() error + Close() error +} + +func newReadCommand(server readCommandServer, mode omode.Mode) *readCommand { + // cat/grep reads are one-shot: their input is exhausted once every file + // has been read to EOF. tail follows its files indefinitely, so its input + // never exhausts and must not finish a output aggregate. + oneShotInput := mode == omode.CatClient || mode == omode.GrepClient return &readCommand{ - server: server, - mode: mode, + server: server, + mode: mode, + shutdownCoordinator: newShutdownCoordinator(server, oneShotInput), } } func (r *readCommand) Start(ctx context.Context, ltx lcontext.LContext, argc int, args []string, retries int) { + r.generation = sessionGenerationFromContext(ctx) re := regex.NewNoop() if argc >= 4 { deserializedRegex, err := regex.Deserialize(strings.Join(args[2:], " ")) if err != nil { - r.server.sendln(r.server.serverMessages, dlog.Server.Error(r.server.user, + r.sendServerMessage(ctx, dlog.Server.Error(r.server.LogContext(), "Unable to parse command", err)) return } re = deserializedRegex } if argc < 3 { - r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user, + r.sendServerMessage(ctx, dlog.Server.Warn(r.server.LogContext(), "Unable to parse command", args, argc)) return } // In serverless mode, can also read data from pipe // e.g.: grep foo bar.log | dmap 'from STATS select ...' - if r.isInputFromPipe() { + // Only read from pipe if no file argument is provided + isPipe := r.isInputFromPipe() && (argc < 2 || args[1] == "" || args[1] == "-") + + if isPipe { dlog.Server.Debug("Reading data from stdin pipe") - // Empty file path and globID "-" represents reading from the stdin pipe. - r.read(ctx, ltx, "", "-", re) + r.readPipe(ctx, ltx, re) + return + } + + if fs.IsJournalSpec(args[1]) { + dlog.Server.Debug("Reading data from journal") + r.readJournal(ctx, ltx, args[1], re, retries) return } @@ -60,38 +89,82 @@ func (r *readCommand) Start(ctx context.Context, ltx lcontext.LContext, r.readGlob(ctx, ltx, args[1], re, retries) } +// readPipe reads the single stdin-pipe input in serverless mode (e.g. +// `grep foo bar.log | dmap 'from STATS select ...'`). Unlike file/glob/journal +// reads it does NOT go through readFiles/readFileIfPermissions, so it must +// reproduce their finalization here: the pipe is accounted as one pending input +// and onFileProcessed is invoked once it drains. That is what drives the +// shutdown coordinator's FinishInput signal to a output aggregate — without it a +// serverless dmap reading from a pipe would block in Aggregate.Start on its +// inputFinished signal forever (the client keeps re-gathering interim results +// and never terminates). Previously the pipe fed a regular Aggregate that +// finalized when its input line channel was closed; the output aggregate has no +// such channel-close, so the pending-file/onFileProcessed path is the +// equivalent "input exhausted" signal. The pending accounting also keeps the +// counter balanced: onFileProcessed unconditionally decrements pendingFiles, so +// it must be paired with an AddPendingFiles here. +func (r *readCommand) readPipe(ctx context.Context, ltx lcontext.LContext, re regex.Regex) { + r.server.AddPendingFiles(1) + defer r.shutdownCoordinator.onFileProcessed("-") + // Empty file path and globID "-" represents reading from the stdin pipe. + r.read(ctx, ltx, "", nil, "-", re) +} + +func (r *readCommand) readJournal(ctx context.Context, ltx lcontext.LContext, + spec string, re regex.Regex, _ int) { + + r.readFiles(ctx, ltx, []string{spec}, spec, re, r.server.ReadGlobRetryInterval()) +} + func (r *readCommand) readGlob(ctx context.Context, ltx lcontext.LContext, glob string, re regex.Regex, retries int) { - retryInterval := time.Second * 5 + retryInterval := r.server.ReadGlobRetryInterval() glob = filepath.Clean(glob) for retryCount := 0; retryCount < retries; retryCount++ { paths, err := filepath.Glob(glob) if err != nil { - dlog.Server.Warn(r.server.user, glob, err) - time.Sleep(retryInterval) + dlog.Server.Warn(r.server.LogContext(), glob, err) + if !ctxutil.Sleep(ctx, retryInterval) { + return + } continue } if numPaths := len(paths); numPaths == 0 { - dlog.Server.Error(r.server.user, "No such file(s) to read", glob) - r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user, + dlog.Server.Error(r.server.LogContext(), "No such file(s) to read", glob) + r.sendServerMessage(ctx, dlog.Server.Warn(r.server.LogContext(), "Unable to read file(s), check server logs")) select { case <-ctx.Done(): return default: } - time.Sleep(retryInterval) + if !ctxutil.Sleep(ctx, retryInterval) { + return + } continue } + // Cap the number of paths to prevent an authenticated user with a broad + // read permission from spawning an unbounded number of goroutines and + // exhausting server memory. Excess paths are dropped with a warning so + // the partial result is still delivered rather than failing entirely. + if cap := r.server.MaxGlobTargets(); len(paths) > cap { + dlog.Server.Warn(r.server.LogContext(), "Glob expansion exceeded cap, truncating", + "glob", glob, "matched", len(paths), "cap", cap) + r.sendServerMessage(ctx, dlog.Server.Warn(r.server.LogContext(), + "Glob expansion exceeded server limit, only first targets served", + "limit", cap, "matched", len(paths))) + paths = paths[:cap] + } + r.readFiles(ctx, ltx, paths, glob, re, retryInterval) return } - r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user, + r.sendServerMessage(ctx, dlog.Server.Warn(r.server.LogContext(), "Giving up to read file(s)")) return } @@ -99,81 +172,255 @@ func (r *readCommand) readGlob(ctx context.Context, ltx lcontext.LContext, func (r *readCommand) readFiles(ctx context.Context, ltx lcontext.LContext, paths []string, glob string, re regex.Regex, retryInterval time.Duration) { + dlog.Server.Info(r.server.LogContext(), "Processing files", "count", len(paths), "glob", glob) + + // Track pending files for this batch + totalPending := r.server.AddPendingFiles(int32(len(paths))) + dlog.Server.Info(r.server.LogContext(), "Added pending files", "count", len(paths), "totalPending", totalPending) + var wg sync.WaitGroup wg.Add(len(paths)) for _, path := range paths { go r.readFileIfPermissions(ctx, ltx, &wg, path, glob, re) } wg.Wait() + + dlog.Server.Info(r.server.LogContext(), "All files processed", "count", len(paths)) + + select { + case <-ctx.Done(): + return + default: + } + + // In output mode, signal EOF once all pending file work is drained. + // Active command count may still include side-effect commands (for example AUTHKEY), + // so relying on "active == 1" can skip EOF signaling and lead to dropped output. + // + // Output is now the only runtime path, so the former config gating on + // the former config gating has been removed. This EOF handshake runs for every + // cat/grep/tail read. + // + // The guard is the mode check rather than Aggregate() == nil: + // SERVER-MODE dmap also enables output mode (readWithProcessor -> + // ensureOutputEnabled) even though its lines feed the Aggregate + // rather than the output channel writer, and it RELIES on this epilogue's + // SignalOutputEOF to disable output mode on the session output goroutine so the + // client receives EOF and the session terminates. Excluding the + // output-aggregate case here (as Aggregate() == nil would) hangs + // server-mode dmap: output mode stays enabled and the client waits forever for + // output that never ends. dmap uses map mode (not cat/grep/tail) so it is + // already excluded from this direct-output epilogue. + // + // For serverless dmap the epilogue is likewise not reached (map mode): the + // output aggregate is finalized through FinishInput in the shutdown + // coordinator, independently of this direct-output handshake. + if r.mode == omode.CatClient || r.mode == omode.GrepClient || r.mode == omode.TailClient { + if r.server.DirectOutputActive() && r.server.HasOutputEOF() { + // Capture the handshake epoch BEFORE the pending check. A command + // that joins the session afterwards bumps the epoch (readFiles + // increments the pending count before its per-file work enables + // output mode), so a joiner invisible to the pending==0 check below + // is guaranteed to advance the epoch after this capture — turning + // our SignalOutputEOF into a no-op instead of cutting off the + // joiner's output mid-batch. The window between the checks below + // and the signal is wide (FlushOutput may block for seconds), + // which is exactly what this guard covers. + // + // Note the never-signaling-joiner class: a output-aggregate (dmap) + // command joining the live session also bumps the epoch via its + // enable, but map mode is excluded from this epilogue and never + // signals EOF itself. Our signal is then dropped and the ack wait + // below runs into its bounded timeout — a spurious warning, but no + // data loss: our data was flushed above and session shutdown + // flushes the rest (see outputManager.signalEOF). + epoch := r.server.OutputEpoch() + + pending, active := r.server.PendingAndActive() + shouldSignalEOF := pending == 0 + if !shouldSignalEOF { + dlog.Server.Trace(r.server.LogContext(), "Skipping output EOF signal for non-final command", + "pending", pending, "active", active) + return + } + + dlog.Server.Debug(r.server.LogContext(), "Output mode: flushing data before EOF signal") + + // Ensure all output data is flushed before signaling EOF. + r.server.FlushOutput() + + // Signal EOF by closing the channel, but only once — and only if + // no newer batch joined since the epoch capture above. + r.server.SignalOutputEOF(epoch) + + // Wait for an explicit reader acknowledgement instead of timing guesses. + if !r.server.Serverless() { + timeout := r.server.OutputEOFAckTimeout() + if r.server.WaitForOutputEOFAck(timeout) { + // The wait is also released when enable() hands the + // handshake over to a new batch (stale refresh), not only + // by a reader ack — the log wording covers both. + dlog.Server.Debug(r.server.LogContext(), "Output EOF handshake released (reader ack or handover)") + // Allow transport buffers to flush after acknowledgement. + if !ctxutil.Sleep(ctx, r.server.ShutdownSerializeWait()) { + return + } + } else { + dlog.Server.Warn( + r.server.LogContext(), + "Timeout waiting for output EOF acknowledgement", + "timeout", timeout, + "remaining", r.server.OutputChannelLen(), + ) + } + } + } + } + + // In output mode with aggregate, we don't close the shared channel here + // because it will be used across multiple invocations + // The aggregate will handle channel closure when it's done } func (r *readCommand) readFileIfPermissions(ctx context.Context, ltx lcontext.LContext, wg *sync.WaitGroup, path, glob string, re regex.Regex) { defer wg.Done() - globID := r.makeGlobID(path, glob) - if !r.server.user.HasFilePermission(path, "readfiles") { - dlog.Server.Error(r.server.user, "No permission to read file", path, globID) - r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user, + defer func() { + r.shutdownCoordinator.onFileProcessed(path) + }() + + globID := r.makeGlobID(ctx, path, glob) + target, ok := r.server.PrepareReadTarget(path) + if !ok { + dlog.Server.Error(r.server.LogContext(), "No permission to read file", path, globID) + r.sendServerMessage(ctx, dlog.Server.Warn(r.server.LogContext(), "Unable to read file(s), check server logs")) return } - r.read(ctx, ltx, path, globID, re) + r.read(ctx, ltx, path, &target, globID, re) } func (r *readCommand) read(ctx context.Context, ltx lcontext.LContext, - path, globID string, re regex.Regex) { + path string, target *fs.ValidatedReadTarget, globID string, re regex.Regex) { + + dlog.Server.Info(r.server.LogContext(), "Start reading", path, globID) + r.logRegexMode(re) - dlog.Server.Info(r.server.user, "Start reading", path, globID) var reader fs.FileReader var limiter chan struct{} + serverMessages, closeServerMessages := r.newGeneratedServerMessagesChannel(ctx) + defer closeServerMessages() switch r.mode { case omode.GrepClient, omode.CatClient: - reader = fs.NewCatFile(path, globID, r.server.serverMessages) - limiter = r.server.catLimiter + if target != nil && target.Kind == fs.JournalKind { + journalReader, err := journal.NewReader(journalArgs(path), path, false, serverMessages) + if err != nil { + r.sendServerMessage(ctx, dlog.Server.Warn(r.server.LogContext(), "Unable to read journal", err)) + return + } + reader = journalReader + } else if target != nil { + catFile := fs.NewValidatedCatFile(path, *target, globID, serverMessages, r.server.MaxLineLength()) + reader = &catFile + } else { + catFile := fs.NewCatFile(path, globID, serverMessages, r.server.MaxLineLength()) + reader = &catFile + } + limiter = r.server.CatLimiter() case omode.TailClient: fallthrough default: - reader = fs.NewTailFile(path, globID, r.server.serverMessages) - limiter = r.server.tailLimiter + if target != nil && target.Kind == fs.JournalKind { + journalReader, err := journal.NewReader(journalArgs(path), path, true, serverMessages) + if err != nil { + r.sendServerMessage(ctx, dlog.Server.Warn(r.server.LogContext(), "Unable to read journal", err)) + return + } + reader = journalReader + } else if target != nil { + tailFile := fs.NewValidatedTailFile(path, *target, globID, serverMessages, r.server.MaxLineLength()) + reader = &tailFile + } else { + tailFile := fs.NewTailFile(path, globID, serverMessages, r.server.MaxLineLength()) + reader = &tailFile + } + limiter = r.server.TailLimiter() } + // acquired tracks whether this goroutine successfully sent to the limiter. + // The defer must only release a slot when this goroutine actually holds one; + // an unconditional release would steal a slot from another goroutine that + // is still holding it, permanently reducing the effective semaphore capacity. + var acquired bool defer func() { - select { - case <-limiter: - default: + if acquired { + <-limiter } }() select { case limiter <- struct{}{}: + acquired = true + dlog.Server.Debug(r.server.LogContext(), "Got limiter slot immediately", "path", path) case <-ctx.Done(): + dlog.Server.Debug(r.server.LogContext(), "Context cancelled while waiting for limiter", "path", path) return default: - dlog.Server.Info("Server limit hit, queueing file", len(limiter), path) + dlog.Server.Info(r.server.LogContext(), "Server limit hit, queueing file", "limiterLen", len(limiter), "path", path, "maxConcurrent", cap(limiter)) select { case limiter <- struct{}{}: - dlog.Server.Info("Server limit OK now, processing file", len(limiter), path) + acquired = true + dlog.Server.Info(r.server.LogContext(), "Server limit OK now, processing file", "limiterLen", len(limiter), "path", path) case <-ctx.Done(): + dlog.Server.Debug(r.server.LogContext(), "Context cancelled while queued for limiter", "path", path) return } } - lines := r.server.lines - aggregate := r.server.aggregate + // Output is the one and only read path. read() is only ever invoked for the + // cat/grep/tail command handlers (see makeReadCommandHandler), and MapReduce + // always builds a Aggregate for both server mode and serverless (see + // mapcommand.go). So either the read runs in cat/grep mode with + // Aggregate() non-nil and feeds it directly via AggregateProcessor, + // or it is genuine cat/grep/tail output using the output direct-output writer; + // makeProcessor picks between the two. The former channel-based fallback + // (readViaChannels feeding the regular server.Aggregate) was removed once + // serverless MapReduce migrated to the output aggregate (tasks sv0/hv0), so + // there is no non-output path left here. + dlog.Server.Debug(r.server.LogContext(), "Selecting read mode", + "mode", r.mode, "hasAggregate", r.server.Aggregate() != nil) + dlog.Server.Info(r.server.LogContext(), "Using turbo mode for reading", path, "mode", r.mode, "hasAggregate", r.server.Aggregate() != nil) + r.readWithProcessor(ctx, ltx, path, globID, re, reader) +} + +func journalArgs(spec string) []string { + source := strings.TrimPrefix(spec, fs.JournalSpecPrefix) + if source == "" { + return nil + } + return []string{"-u", source} +} + +func (r *readCommand) readWithProcessor(ctx context.Context, ltx lcontext.LContext, + path, globID string, re regex.Regex, reader fs.FileReader) { + + dlog.Server.Info(r.server.LogContext(), "Using output channel-less implementation", path, globID) + r.logRegexMode(re) + + r.ensureOutputEnabled(ctx) + writer := r.makeWriter(ctx) + + r.executeReadLoop(ctx, ltx, path, globID, re, reader, r.readViaProcessor(path, globID, writer)) +} + +func (r *readCommand) executeReadLoop(ctx context.Context, ltx lcontext.LContext, + path, globID string, re regex.Regex, reader fs.FileReader, strategy readStrategy) { for { - if aggregate != nil { - lines = make(chan *line.Line, 100) - aggregate.NextLinesCh <- lines - } - if err := reader.Start(ctx, ltx, lines, re); err != nil { - dlog.Server.Error(r.server.user, path, globID, err) - } - if aggregate != nil { - // Also makes aggregate to Flush - close(lines) + if err := strategy(ctx, ltx, reader, re); err != nil { + dlog.Server.Error(r.server.LogContext(), path, globID, err) } select { @@ -184,12 +431,135 @@ func (r *readCommand) read(ctx context.Context, ltx lcontext.LContext, return } } - time.Sleep(time.Second * 2) + + if !ctxutil.Sleep(ctx, r.server.ReadRetryInterval()) { + return + } dlog.Server.Info(path, globID, "Reading file again") } } -func (r *readCommand) makeGlobID(path, glob string) string { +func (r *readCommand) readViaProcessor(path, globID string, writer LineWriter) readStrategy { + return func(ctx context.Context, ltx lcontext.LContext, reader fs.FileReader, re regex.Regex) error { + dlog.Server.Trace(r.server.LogContext(), path, globID, "readWithProcessor -> starting read loop iteration") + + processor := r.makeProcessor(path, globID, writer) + + dlog.Server.Trace(r.server.LogContext(), path, globID, "readWithProcessor -> reader.StartWithPocessorOptimized -> about to start") + startErr := reader.StartWithProcessorOptimized(ctx, ltx, processor, re) + dlog.Server.Trace(r.server.LogContext(), path, globID, "readWithProcessor -> reader.StartWithPocessorOptimized -> completed") + + // Ensure we flush and close the processor before retry checks. + dlog.Server.Trace(r.server.LogContext(), path, globID, "readWithProcessor -> flushing processor") + if flushErr := processor.Flush(); flushErr != nil { + dlog.Server.Error(r.server.LogContext(), path, globID, "flush error", flushErr) + } + dlog.Server.Trace(r.server.LogContext(), path, globID, "readWithProcessor -> closing processor") + if closeErr := processor.Close(); closeErr != nil { + dlog.Server.Error(r.server.LogContext(), path, globID, "close error", closeErr) + } + dlog.Server.Trace(r.server.LogContext(), path, globID, "readWithProcessor -> processor closed") + + // Give time for data to be transmitted. + // This is crucial for integration tests to ensure all data is sent + // Skip this delay in serverless mode since data is written directly to stdout + if !r.server.Serverless() { + dlog.Server.Trace(r.server.LogContext(), path, globID, "readWithProcessor -> waiting for data transmission") + if !ctxutil.Sleep(ctx, r.server.OutputTransmissionDelay()) { + return startErr + } + } + + return startErr + } +} + +func (r *readCommand) ensureOutputEnabled(ctx context.Context) { + // EnableDirectOutput is an atomic check-and-enable guarded by the output + // manager's mutex, so mode and channel initialization are always observed + // together — no visibility double-check is needed here. It returns false + // when output mode was already active. + if !r.server.EnableDirectOutput() { + return + } + // Wake a potentially blocked reader goroutine so it can switch to output drain path. + r.sendServerMessage(ctx, ".output wake") +} + +func (r *readCommand) makeWriter(ctx context.Context) LineWriter { + // Create a writer instance per file to keep concurrent processing isolated. + if r.server.Serverless() { + return NewGeneratedDirectWriter(serverlessOutputWriter(), r.server.Hostname(), r.server.PlainOutput(), r.server.Serverless(), r.generation, r.server.ActiveSessionGeneration) + } + + // Use NewNetworkWriter so bufSize is set to 64KB. A bare struct literal + // here previously left bufSize at zero, which disabled write batching and + // sent every line as its own output-channel payload (one SSH packet + one + // write syscall per line), making server-mode output output far slower than + // it should be. + return NewNetworkWriter(ctx, r.server.GetOutputChannel(), + r.server.ServerMessagesChannel(), r.server.Hostname(), + r.server.PlainOutput(), r.server.Serverless(), r.generation, + r.server.ActiveSessionGeneration) +} + +// serverlessOutputWriter returns the io.Writer the serverless direct-output +// (output) path writes payload to. It is os.Stdout by default. When the client +// opted in via --log-payload / Client.LogPayload, it additionally tees the exact +// same payload bytes into the fout daily log FILE sink. +// +// This is needed because output is now the only runtime path: the serverless +// direct-output path writes payload straight to stdout and bypasses the fout +// logger's Raw method, so the logger's own --log-payload file tee never runs. +// The tee is added via io.MultiWriter, which writes to os.Stdout first and byte +// for byte unchanged, so stdout stays identical whether or not --log-payload is +// set; only the file gains the payload. When LogPayload is off (the default) we +// return the bare os.Stdout so the hot path stays allocation-free. +func serverlessOutputWriter() io.Writer { + if config.Client != nil && config.Client.LogPayload { + return io.MultiWriter(os.Stdout, payloadFileTeeWriter{}) + } + return os.Stdout +} + +// payloadFileTeeWriter is an io.Writer that forwards serverless payload bytes to +// the client logger's FILE sink only (never stdout), honoring --log-payload. It +// lets the serverless direct-output path reuse the fout daily-log file tee that +// the bypassed logger.Raw path would otherwise have provided. +type payloadFileTeeWriter struct{} + +func (payloadFileTeeWriter) Write(p []byte) (int, error) { + // dlog.Client is the fout logger that owns the daily log file in serverless + // mode; RawPayloadFileTee no-ops when the logger has no file sink or payload + // teeing is disabled. Report the full length as written so io.MultiWriter + // does not treat the tee as a short write. + if dlog.Client != nil { + dlog.Client.RawPayloadFileTee(string(p)) + } + return len(p), nil +} + +func (r *readCommand) makeProcessor(path, globID string, writer LineWriter) readProcessor { + if aggregate := r.server.Aggregate(); aggregate != nil { + dlog.Server.Info(r.server.LogContext(), "Using turbo aggregate processor for MapReduce", path, globID) + return server.NewAggregateProcessor(aggregate, globID) + } + + return NewDirectLineProcessor(writer, globID) +} + +func (r *readCommand) logRegexMode(re regex.Regex) { + if r.mode != omode.GrepClient { + return + } + if re.IsLiteral() { + dlog.Server.Info(r.server.LogContext(), "Using optimized literal string matching for pattern:", re.Pattern()) + } else { + dlog.Server.Info(r.server.LogContext(), "Using regex matching for pattern:", re.Pattern()) + } +} + +func (r *readCommand) makeGlobID(ctx context.Context, path, glob string) string { var idParts []string pathParts := strings.Split(path, "/") @@ -206,13 +576,54 @@ func (r *readCommand) makeGlobID(path, glob string) string { return pathParts[len(pathParts)-1] } - r.server.sendln(r.server.serverMessages, - dlog.Server.Warn("Empty file path given?", path, glob)) + r.sendServerMessage(ctx, dlog.Server.Warn("Empty file path given?", path, glob)) return "" } +// sendServerMessage forwards a user-visible message to the session's shared +// serverMessages channel (capacity 10, drained only by baseHandler.Read). A +// bare send here would block forever once the client disconnects and Read +// stops draining — pinning one goroutine per stuck send (e.g. one per +// permission-denied file of a glob expansion, up to MaxGlobTargets). The +// select on ctx.Done() makes the send abandonable: the per-command context is +// cancelled on command completion and on handler shutdown (see +// baseHandler.newCommandContext), mirroring the done-guarded baseHandler.send. +func (r *readCommand) sendServerMessage(ctx context.Context, message string) { + select { + case r.server.ServerMessagesChannel() <- encodeGeneratedMessage(r.generation, message+"\n"): + case <-ctx.Done(): + } +} + +func (r *readCommand) newGeneratedServerMessagesChannel(ctx context.Context) (chan string, func()) { + serverMessages := make(chan string, 16) + done := make(chan struct{}) + go func() { + defer close(done) + for { + select { + case message, ok := <-serverMessages: + if !ok { + return + } + select { + case r.server.ServerMessagesChannel() <- encodeGeneratedMessage(r.generation, message): + case <-ctx.Done(): + return + } + case <-ctx.Done(): + return + } + } + }() + return serverMessages, func() { + close(serverMessages) + <-done + } +} + func (r *readCommand) isInputFromPipe() bool { - if !r.server.serverless { + if !r.server.Serverless() { // Can read from pipe only in serverless mode. return false } |