weaviate/weaviate · error

open change-log stream on %s: %w

Error message

open change-log stream on %s: %w

What it means

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.

Source

Thrown at cluster/replication/copier/copier_changelog.go:67

func (c *Copier) TailAndApply(ctx context.Context, srcNodeId, indexName, shardName, opID string, untilLSN uint64) (lastAppliedLSN uint64, err error) {
	client, err := c.dialSource(ctx, srcNodeId)
	if err != nil {
		return 0, err
	}

	index := c.dbWrapper.GetIndex(schema.ClassName(indexName))
	if index == nil {
		return 0, fmt.Errorf("local index %q not found", indexName)
	}

	stream, err := client.GetChangeLog(ctx, &protocol.GetChangeLogRequest{
		IndexName: indexName,
		ShardName: shardName,
		OpId:      opID,
		UntilLsn:  untilLSN,
	})
	if err != nil {
		return 0, fmt.Errorf("open change-log stream on %s: %w", srcNodeId, err)
	}

	apply := func(ctx context.Context, batch []db.ChangeLogReplayEntry) error {
		return index.OverwriteObjectsFromChangeLog(ctx, shardName, batch)
	}
	return changelogdrain.Drain(ctx, stream, apply)
}

// SnapshotChangeLogLSN returns the source's current change-log LSN without
// sealing it. The log stays writable; pair with a capped TailAndApply to
// drain a phase boundary without sealing.
func (c *Copier) SnapshotChangeLogLSN(ctx context.Context, srcNodeId, indexName, shardName, opID string) (uint64, error) {
	client, err := c.dialSource(ctx, srcNodeId)
	if err != nil {
		return 0, err
	}
	resp, err := client.SnapshotChangeLogLSN(ctx, &protocol.SnapshotChangeLogLSNRequest{
		IndexName: indexName,

View on GitHub (pinned to 75aa4b6d11)

Solutions

1. Check the wrapped cause (%w) to distinguish network vs RPC-level rejection
2. Verify the source node is up and its gRPC port is reachable (dialSource resolves it via nodeSelector)
3. Re-run the replication op so StartChangeCapture re-registers the opID on the source before tailing
4. Confirm the source shard is loaded and the op exists on the source; cancel and re-create the replication op if stale
5. Check both nodes run compatible Weaviate versions supporting the change-log replication RPCs

Example fix

// before: failing when source is temporarily unreachable
lsn, err := copier.TailAndApply(ctx, index, shard, srcNode, opID, untilLSN)
// after: retry transient open failures
err := backoff.Retry(func() error {
    var err error
    lsn, err = copier.TailAndApply(ctx, index, shard, srcNode, opID, untilLSN)
    return err
}, backoff.WithContext(backoff.NewExponentialBackOff(), ctx))

// verify source node is a reachable, compatible member before tailing
if _, err := nodeSelector.NodeGRPCPort(srcNodeId); err != nil {
    return fmt.Errorf("source %s has no gRPC endpoint: %w", srcNodeId, err)
}
if err := pingGRPC(ctx, srcNodeId); err != nil {
    return fmt.Errorf("source %s unreachable: %w", srcNodeId, err)
}

func canTailSource(sel NodeSelector, srcNodeId string) bool {
    _, err := sel.NodeGRPCPort(srcNodeId)
    return err == nil
}

var lsn uint64
err := backoff.Retry(func() error {
    var e error
    lsn, e = copier.TailAndApply(ctx, index, shard, srcNode, opID, untilLSN)
    if e != nil && isTerminalOpError(e) {
        return backoff.Permanent(e)
    }
    return e
}, backoff.WithContext(backoff.NewExponentialBackOff(), ctx))
if err != nil {
    logger.Errorf("tail+apply failed for op %d: %v", opID, err)
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• failed to send GetFile request for %s: %w
• failed to receive file chunk for %s: %w
• recv change-log entry: %w
• read digest: %w
• get gRPC connection to %s: %w

AI-assisted analysis of weaviate/weaviate@75aa4b6d11 (2026-09-04). Data as JSON: /api/errors/939402c49dc17d38. Report an issue: GitHub.