{"record":{"id":"939402c49dc17d38","repo":"weaviate/weaviate","slug":"open-change-log-stream-on-s-w","errorCode":null,"errorMessage":"open change-log stream on %s: %w","messagePattern":"open change-log stream on (.+?): %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"cluster/replication/copier/copier_changelog.go","lineNumber":67,"sourceCode":"func (c *Copier) TailAndApply(ctx context.Context, srcNodeId, indexName, shardName, opID string, untilLSN uint64) (lastAppliedLSN uint64, err error) {\n\tclient, err := c.dialSource(ctx, srcNodeId)\n\tif err != nil {\n\t\treturn 0, err\n\t}\n\n\tindex := c.dbWrapper.GetIndex(schema.ClassName(indexName))\n\tif index == nil {\n\t\treturn 0, fmt.Errorf(\"local index %q not found\", indexName)\n\t}\n\n\tstream, err := client.GetChangeLog(ctx, &protocol.GetChangeLogRequest{\n\t\tIndexName: indexName,\n\t\tShardName: shardName,\n\t\tOpId:      opID,\n\t\tUntilLsn:  untilLSN,\n\t})\n\tif err != nil {\n\t\treturn 0, fmt.Errorf(\"open change-log stream on %s: %w\", srcNodeId, err)\n\t}\n\n\tapply := func(ctx context.Context, batch []db.ChangeLogReplayEntry) error {\n\t\treturn index.OverwriteObjectsFromChangeLog(ctx, shardName, batch)\n\t}\n\treturn changelogdrain.Drain(ctx, stream, apply)\n}\n\n// SnapshotChangeLogLSN returns the source's current change-log LSN without\n// sealing it. The log stays writable; pair with a capped TailAndApply to\n// drain a phase boundary without sealing.\nfunc (c *Copier) SnapshotChangeLogLSN(ctx context.Context, srcNodeId, indexName, shardName, opID string) (uint64, error) {\n\tclient, err := c.dialSource(ctx, srcNodeId)\n\tif err != nil {\n\t\treturn 0, err\n\t}\n\tresp, err := client.SnapshotChangeLogLSN(ctx, &protocol.SnapshotChangeLogLSNRequest{\n\t\tIndexName: indexName,","sourceCodeStart":49,"sourceCodeEnd":85,"githubUrl":"https://github.com/weaviate/weaviate/blob/75aa4b6d11f8818305aafd4440b4e32794f7ca04/cluster/replication/copier/copier_changelog.go#L49-L85","documentation":"TailAndApply opens a change-log stream from the source node via gRPC (StartChangeCapture/dialSource path) and wraps any failure of that stream-open RPC with this message. It means the copier could not establish or open the change-log stream for the given index/shard on the source node. The underlying cause (network, RPC rejection, shard not loaded) is preserved via %w.","triggerScenarios":"Calling TailAndApply(ctx, indexName, shardName, srcNodeId, opID, untilLSN) when the source node is unreachable, the gRPC stream open RPC returns an error (stream refused, opID unknown/rejected, source shard not loaded), or the underlying client.SnapshotChangeLog/stream call fails mid-handshake.","commonSituations":"Source node restarting or down during replica replication; network partition between nodes; the opID was never registered on the source (e.g. StartChangeCapture failed earlier); source shard was dropped or not yet loaded on the source; version mismatch where the peer does not implement the change-log RPC.","solutions":["Check the wrapped cause (%w) to distinguish network vs RPC-level rejection","Verify the source node is up and its gRPC port is reachable (dialSource resolves it via nodeSelector)","Re-run the replication op so StartChangeCapture re-registers the opID on the source before tailing","Confirm the source shard is loaded and the op exists on the source; cancel and re-create the replication op if stale","Check both nodes run compatible Weaviate versions supporting the change-log replication RPCs"],"exampleFix":"// before: failing when source is temporarily unreachable\nlsn, err := copier.TailAndApply(ctx, index, shard, srcNode, opID, untilLSN)\n// after: retry transient open failures\nerr := backoff.Retry(func() error {\n    var err error\n    lsn, err = copier.TailAndApply(ctx, index, shard, srcNode, opID, untilLSN)\n    return err\n}, backoff.WithContext(backoff.NewExponentialBackOff(), ctx))","handlingStrategy":"retry","validationCode":"// verify source node is a reachable, compatible member before tailing\nif _, err := nodeSelector.NodeGRPCPort(srcNodeId); err != nil {\n    return fmt.Errorf(\"source %s has no gRPC endpoint: %w\", srcNodeId, err)\n}\nif err := pingGRPC(ctx, srcNodeId); err != nil {\n    return fmt.Errorf(\"source %s unreachable: %w\", srcNodeId, err)\n}","typeGuard":"func canTailSource(sel NodeSelector, srcNodeId string) bool {\n    _, err := sel.NodeGRPCPort(srcNodeId)\n    return err == nil\n}","tryCatchPattern":"var lsn uint64\nerr := backoff.Retry(func() error {\n    var e error\n    lsn, e = copier.TailAndApply(ctx, index, shard, srcNode, opID, untilLSN)\n    if e != nil && isTerminalOpError(e) {\n        return backoff.Permanent(e)\n    }\n    return e\n}, backoff.WithContext(backoff.NewExponentialBackOff(), ctx))\nif err != nil {\n    logger.Errorf(\"tail+apply failed for op %d: %v\", opID, err)\n}","preventionTips":["Pre-flight check node membership and gRPC port before starting replication ops","Keep both nodes on the same Weaviate version","Monitor inter-node connectivity with health probes","Ensure StartChangeCapture succeeds before tailing the stream"],"tags":["grpc","replication","network","streaming"],"backgroundTag":"grpc-stream-open-failed","analyzedSha":"75aa4b6d11f8818305aafd4440b4e32794f7ca04","analyzedAt":"2026-09-04T14:58:20.392Z","contentChangedAt":"2026-09-04T14:58:20.392Z","schemaVersion":2},"datasetVersion":"2026-09-11T21:17:09.523Z"}