{"record":{"id":"062ce1a1d8bc0b6a","repo":"apache/beam","slug":"statechannel-v-send-v-channel-closed-due-to-v","errorCode":null,"errorMessage":"StateChannel[%v].Send(%v): channel closed due to: %v","messagePattern":"StateChannel\\[(.+?)\\]\\.Send\\((.+?)\\): channel closed due to: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/harness/statemgr.go","lineNumber":786,"sourceCode":"\tc.mu.Unlock()\n\t// Clean up everything else, this stream is done.\n\tc.terminateStreamOnError(err)\n\n\tif ok {\n\t\tch <- &fnpb.StateResponse{Id: id, Error: fmt.Sprintf(\"StateChannel[%v].write failed to send: %v\", c.id, err)}\n\t}\n}\n\n// Send sends a state request and returns the response.\nfunc (c *StateChannel) Send(req *fnpb.StateRequest) (*fnpb.StateResponse, error) {\n\tid := fmt.Sprintf(\"r%v\", atomic.AddInt32(&c.nextRequestNo, 1))\n\treq.Id = id\n\n\tch := make(chan *fnpb.StateResponse, 1)\n\tc.mu.Lock()\n\tif c.closedErr != nil {\n\t\tdefer c.mu.Unlock()\n\t\treturn nil, errors.Wrapf(c.closedErr, \"StateChannel[%v].Send(%v): channel closed due to: %v\", c.id, id, c.closedErr)\n\t}\n\tc.responses[id] = ch\n\tc.mu.Unlock()\n\n\tc.requests <- req\n\n\tvar resp *fnpb.StateResponse\n\tselect {\n\tcase resp = <-ch:\n\tcase <-c.DoneCh:\n\t\tc.mu.Lock()\n\t\tdefer c.mu.Unlock()\n\t\treturn nil, errors.Wrapf(c.closedErr, \"StateChannel[%v].Send(%v): context canceled\", c.id, id)\n\t}\n\tif resp.Error != \"\" {\n\t\treturn nil, errors.New(resp.Error)\n\t}\n\treturn resp, nil","sourceCodeStart":768,"sourceCodeEnd":804,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/harness/statemgr.go#L768-L804","documentation":"StateChannel.Send fails when the channel's closedErr is set — the underlying gRPC stream to the state service has terminated (recv loop got an error/EOF and closed the channel). Send refuses new requests and reports the original close reason.","triggerScenarios":"Any state read/write request issued after the channel's receive loop ended: runner closed the stream, network error, connection reset, or the server sent an error causing channel close.","commonSituations":"Long-running bundles where the state stream times out or is dropped, runner restarts mid-bundle, network instability between worker and runner, concurrent close during shutdown racing a send.","solutions":["Inspect the wrapped closedErr for the root cause (EOF, deadline, reset) and address that first.","Check network stability and any load-balancer/proxy idle timeouts between worker and runner; raise idle timeouts.","Reduce bundle duration or state request volume if the runner is dropping overloaded streams.","Retry the bundle/job; if the runner crashed, scale or fix the runner and rerun."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"resp, err := w.Append(ctx, val)\nif err != nil && strings.Contains(err.Error(), \"channel closed due to\") {\n    return fmt.Errorf(\"state stream dropped mid-bundle: %w\", err) // surface for runner-level retry\n}","preventionTips":["Raise gRPC keepalive/idle timeouts on intermediaries","Keep bundles short to limit stream lifetime","Detect runner restarts and fail the bundle early"],"tags":["go","grpc","network","beam","channel-closed"],"backgroundTag":"broken-pipe","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}