#347

Committed 31 Jul 2026 06:59PM UTC coverage: 68.148% (-0.03%) from 68.181%

Build # #347

Build Type

push

github

Committed by

web-flow

Commit Message

NEXUS-485: Support Workflow Update as a Nexus Operation (#2945)

* NEXUS-485: Support Workflow Update as a Nexus Operation

* address comments, change signatures to newer

* address comments 2: add all workflow exec overloads

* address comments: log failed

Coverage Stats

7166 of 12554 branches covered (57.08%)

Branch coverage included in aggregate %.

118 of 296 new or added lines in 11 files covered. (39.86%)

28 existing lines in 7 files now uncovered.

29722 of 41575 relevant lines covered (71.49%)

0.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.89

/temporal-sdk/src/main/java/io/temporal/internal/payload/visitor/Traversal.java

package io.temporal.internal.payload.visitor;

import com.google.protobuf.Any;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.Message;
import com.google.protobuf.MessageOrBuilder;
import io.temporal.api.common.v1.Payload;
import io.temporal.internal.common.AsyncSemaphore;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;

/**
 * Mutable state for one traversal, called into by the generated per-message visitors.
 *
 * <p>The single-threaded walk only records a job and a write-back per payload sequence; {@link
 * #execute()} runs the visits and applies the write-backs afterward in walk order, so the
 * non-thread-safe builders are never mutated concurrently. Visits are asynchronous, so the engine
 * needs no executor — it only bounds how many of their futures are outstanding.
 */
final class Traversal {
  // Null for a message-only traversal: payload seams are skipped, only the MessageVisitor fires.
  private final PayloadVisitor<Object> payloadVisitor;
  private final MessageVisitor<Object> messageVisitor;
  private final Map<String, MessageRegistryEntry> registry;
  final boolean skipSearchAttributes;
  final boolean skipHeaders;
  private final int concurrency;

  private final List<LeafJob> jobs = new ArrayList<>();
  private final List<Runnable> writeBacks = new ArrayList<>();
  private Object currentContext;

  @SuppressWarnings("unchecked")
  Traversal(
      PayloadVisitor<?> payloadVisitor,
      MessageVisitor<?> messageVisitor,
      Object initialContext,
      boolean skipSearchAttributes,
      boolean skipHeaders,
      int concurrency,
      Map<String, MessageRegistryEntry> registry) {
    if (concurrency < 1) {
      throw new IllegalArgumentException("concurrency must be at least 1, got " + concurrency);
    }
    this.payloadVisitor = (PayloadVisitor<Object>) payloadVisitor;
    this.messageVisitor = (MessageVisitor<Object>) messageVisitor;
    this.currentContext = initialContext;
    this.skipSearchAttributes = skipSearchAttributes;
    this.skipHeaders = skipHeaders;
    this.concurrency = concurrency;
    this.registry = registry;
  }

  // --- Structural walk: called by generated code ---

  /** No-op for a type with no payloads. */
  void dispatch(Message.Builder builder) {
    MessageRegistryEntry entry = registry.get(builder.getDescriptorForType().getFullName());
    if (entry != null) {
      entry.visitor.visit(this, builder);
    }
  }

  /** Narrows the scoped context; returns the value {@link #exit} restores. */
  Object enter(MessageOrBuilder message) {
    Object previous = currentContext;
    if (messageVisitor != null) {
      currentContext = messageVisitor.onEnter(previous, message);
    }
    return previous;
  }

  void exit(Object previous) {
    currentContext = previous;
  }

  /** Record a visit of a payload sequence (a {@code Payloads} or {@code repeated Payload}). */
  void payloads(List<Payload> batch, Consumer<List<Payload>> writeBack) {
    if (payloadVisitor == null) {
      return;
    }
    LeafJob job = new LeafJob(batch, currentContext, false);
    jobs.add(job);
    writeBacks.add(() -> writeBack.accept(job.result));
  }

  /** The visitor must return exactly one payload (checked in {@link #record}). */
  void singlePayload(Payload value, Consumer<Payload> writeBack) {
    if (payloadVisitor == null) {
      return;
    }
    LeafJob job = new LeafJob(Collections.singletonList(value), currentContext, true);
    jobs.add(job);
    writeBacks.add(() -> writeBack.accept(job.result.get(0)));
  }

  /** Applied after all visits, single-threaded, in walk order. */
  void deferWriteBack(Runnable writeBack) {
    writeBacks.add(writeBack);
  }

  /** Unpack a {@code google.protobuf.Any}, traverse its contents, and re-pack after visits. */
  void any(Any.Builder anyBuilder) {
    String typeUrl = anyBuilder.getTypeUrl();
    int slash = typeUrl.lastIndexOf('/');
    String fullName = slash >= 0 ? typeUrl.substring(slash + 1) : typeUrl;
    MessageRegistryEntry entry = registry.get(fullName);
    if (entry == null) {
      // Unknown or payload-free type: leave the Any untouched.
      return;
    }
    Message.Builder inner = entry.newBuilder.get();
    try {
      inner.mergeFrom(anyBuilder.getValue());
    } catch (InvalidProtocolBufferException e) {
      throw new VisitorException("failed to unpack Any of type " + fullName, e);
    }
    entry.visitor.visit(this, inner);
    deferWriteBack(() -> anyBuilder.setValue(inner.build().toByteString()));
  }

  // --- Execution: visits, then write-backs ---

  /**
   * Completes the returned future once the visits and write-backs are done. Blocks no thread of its
   * own: the caller decides how to wait and which executor to chain on. Write-backs run on whatever
   * thread completes the last visit (inline if the visits are synchronous). A visit failure aborts
   * the traversal — remaining visits unstarted, write-backs skipped — and completes the future
   * exceptionally with the original throwable.
   */
  CompletableFuture<Void> execute() {
    CompletableFuture<Void> visitsDone =
        jobs.isEmpty() ? CompletableFuture.completedFuture(null) : runVisits();
    CompletableFuture<Void> result = new CompletableFuture<>();
    visitsDone.whenComplete(
        (v, err) -> {
          if (err != null) {
            result.completeExceptionally(unwrap(err));
            return;
          }
          try {
            for (Runnable writeBack : writeBacks) {
              writeBack.run();
            }
            result.complete(null);
          } catch (Throwable t) {
            result.completeExceptionally(t);
          }
        });
    return result;
  }

  /** Run the visits with at most {@code concurrency} outstanding; fails with the first error. */
  private CompletableFuture<Void> runVisits() {
    AsyncSemaphore permits = new AsyncSemaphore(concurrency);
    AtomicReference<Throwable> firstError = new AtomicReference<>();
    List<CompletableFuture<Void>> all = new ArrayList<>(jobs.size());
    for (LeafJob job : jobs) {
      all.add(permits.acquire().thenCompose(p -> runJob(job, permits, firstError)));
    }
    return CompletableFuture.allOf(all.toArray(new CompletableFuture[0]))
        .thenCompose(
            v -> {
              Throwable error = firstError.get();
              if (error == null) {
                return CompletableFuture.completedFuture(null);
              }
              CompletableFuture<Void> failed = new CompletableFuture<>();
              failed.completeExceptionally(error);
              return failed;
            });
  }

  /** Runs one job under a held permit, releasing it exactly once when the visit settles. */
  private CompletableFuture<Void> runJob(
      LeafJob job, AsyncSemaphore permits, AtomicReference<Throwable> firstError) {
    if (firstError.get() != null) {
      permits.release();
      return CompletableFuture.completedFuture(null);
    }
    CompletableFuture<List<Payload>> visit;
    try {
      visit = payloadVisitor.visit(job.context, job.input);
    } catch (RuntimeException | Error t) {
      firstError.compareAndSet(null, t);
      permits.release();
      return CompletableFuture.completedFuture(null);
    }
    if (visit == null) {
      firstError.compareAndSet(null, new IllegalStateException("payload visitor returned null"));
      permits.release();
      return CompletableFuture.completedFuture(null);
    }
    return visit
        .handle(
            (result, err) -> {
              record(job, result, err, firstError);
              return (Void) null;
            })
        .whenComplete((v, e) -> permits.release());
  }

  private void record(
      LeafJob job, List<Payload> result, Throwable err, AtomicReference<Throwable> firstError) {
    if (err != null) {
      firstError.compareAndSet(null, unwrap(err));
    } else if (result == null) {
      firstError.compareAndSet(null, new IllegalStateException("payload visitor returned null"));
    } else if (job.single && result.size() != 1) {
      firstError.compareAndSet(
          null,
          new IllegalStateException(
              "single-payload field requires exactly 1 returned payload, got " + result.size()));
    } else {
      job.result = result;
    }
  }

  /** Strip the {@link CompletionException} a dependent stage wraps around its cause. */
  private static Throwable unwrap(Throwable t) {
    return (t instanceof CompletionException && t.getCause() != null) ? t.getCause() : t;
  }

  /** A recorded visit and the slot its result lands in. */
  private static final class LeafJob {
    final List<Payload> input;
    final Object context;
    final boolean single;
    volatile List<Payload> result;

    LeafJob(List<Payload> input, Object context, boolean single) {
      this.input = input;
      this.context = context;
      this.single = single;
    }
  }
}

1	package io.temporal.internal.payload.visitor;
2
3	import com.google.protobuf.Any;
4	import com.google.protobuf.InvalidProtocolBufferException;
5	import com.google.protobuf.Message;
6	import com.google.protobuf.MessageOrBuilder;
7	import io.temporal.api.common.v1.Payload;
8	import io.temporal.internal.common.AsyncSemaphore;
9	import java.util.ArrayList;
10	import java.util.Collections;
11	import java.util.List;
12	import java.util.Map;
13	import java.util.concurrent.CompletableFuture;
14	import java.util.concurrent.CompletionException;
15	import java.util.concurrent.atomic.AtomicReference;
16	import java.util.function.Consumer;
17
18	/**
19	* Mutable state for one traversal, called into by the generated per-message visitors.
20	*
21	* <p>The single-threaded walk only records a job and a write-back per payload sequence; {@link
22	* #execute()} runs the visits and applies the write-backs afterward in walk order, so the
23	* non-thread-safe builders are never mutated concurrently. Visits are asynchronous, so the engine
24	* needs no executor — it only bounds how many of their futures are outstanding.
25	*/
26	final class Traversal {
27	// Null for a message-only traversal: payload seams are skipped, only the MessageVisitor fires.
28	private final PayloadVisitor<Object> payloadVisitor;
29	private final MessageVisitor<Object> messageVisitor;
30	private final Map<String, MessageRegistryEntry> registry;
31	final boolean skipSearchAttributes;
32	final boolean skipHeaders;
33	private final int concurrency;
34
35	private final List<LeafJob> jobs = new ArrayList<>();	1✔
36	private final List<Runnable> writeBacks = new ArrayList<>();	1✔
37	private Object currentContext;
38
39	@SuppressWarnings("unchecked")
40	Traversal(
41	PayloadVisitor<?> payloadVisitor,
42	MessageVisitor<?> messageVisitor,
43	Object initialContext,
44	boolean skipSearchAttributes,
45	boolean skipHeaders,
46	int concurrency,
47	Map<String, MessageRegistryEntry> registry) {	1✔
48	if (concurrency < 1) {	1!
49	throw new IllegalArgumentException("concurrency must be at least 1, got " + concurrency);	×
50	}
51	this.payloadVisitor = (PayloadVisitor<Object>) payloadVisitor;	1✔
52	this.messageVisitor = (MessageVisitor<Object>) messageVisitor;	1✔
53	this.currentContext = initialContext;	1✔
54	this.skipSearchAttributes = skipSearchAttributes;	1✔
55	this.skipHeaders = skipHeaders;	1✔
56	this.concurrency = concurrency;	1✔
57	this.registry = registry;	1✔
58	}	1✔
59
60	// --- Structural walk: called by generated code ---
61
62	/** No-op for a type with no payloads. */
63	void dispatch(Message.Builder builder) {
64	MessageRegistryEntry entry = registry.get(builder.getDescriptorForType().getFullName());	1✔
65	if (entry != null) {	1!
66	entry.visitor.visit(this, builder);	1✔
67	}
68	}	1✔
69
70	/** Narrows the scoped context; returns the value {@link #exit} restores. */
71	Object enter(MessageOrBuilder message) {
72	Object previous = currentContext;	1✔
73	if (messageVisitor != null) {	1✔
74	currentContext = messageVisitor.onEnter(previous, message);	1✔
75	}
76	return previous;	1✔
77	}
78
79	void exit(Object previous) {
80	currentContext = previous;	1✔
81	}	1✔
82
83	/** Record a visit of a payload sequence (a {@code Payloads} or {@code repeated Payload}). */
84	void payloads(List<Payload> batch, Consumer<List<Payload>> writeBack) {
85	if (payloadVisitor == null) {	1✔
86	return;	1✔
87	}
88	LeafJob job = new LeafJob(batch, currentContext, false);	1✔
89	jobs.add(job);	1✔
90	writeBacks.add(() -> writeBack.accept(job.result));	1✔
91	}	1✔
92
93	/** The visitor must return exactly one payload (checked in {@link #record}). */
94	void singlePayload(Payload value, Consumer<Payload> writeBack) {
95	if (payloadVisitor == null) {	1✔
96	return;	1✔
97	}
98	LeafJob job = new LeafJob(Collections.singletonList(value), currentContext, true);	1✔
99	jobs.add(job);	1✔
100	writeBacks.add(() -> writeBack.accept(job.result.get(0)));	1✔
101	}	1✔
102
103	/** Applied after all visits, single-threaded, in walk order. */
104	void deferWriteBack(Runnable writeBack) {
105	writeBacks.add(writeBack);	1✔
106	}	1✔
107
108	/** Unpack a {@code google.protobuf.Any}, traverse its contents, and re-pack after visits. */
109	void any(Any.Builder anyBuilder) {
110	String typeUrl = anyBuilder.getTypeUrl();	1✔
111	int slash = typeUrl.lastIndexOf('/');	1✔
112	String fullName = slash >= 0 ? typeUrl.substring(slash + 1) : typeUrl;	1!
113	MessageRegistryEntry entry = registry.get(fullName);	1✔
114	if (entry == null) {	1✔
115	// Unknown or payload-free type: leave the Any untouched.
116	return;	1✔
117	}
118	Message.Builder inner = entry.newBuilder.get();	1✔
119	try {
120	inner.mergeFrom(anyBuilder.getValue());	1✔
121	} catch (InvalidProtocolBufferException e) {	×
122	throw new VisitorException("failed to unpack Any of type " + fullName, e);	×
123	}	1✔
124	entry.visitor.visit(this, inner);	1✔
125	deferWriteBack(() -> anyBuilder.setValue(inner.build().toByteString()));	1✔
126	}	1✔
127
128	// --- Execution: visits, then write-backs ---
129
130	/**
131	* Completes the returned future once the visits and write-backs are done. Blocks no thread of its
132	* own: the caller decides how to wait and which executor to chain on. Write-backs run on whatever
133	* thread completes the last visit (inline if the visits are synchronous). A visit failure aborts
134	* the traversal — remaining visits unstarted, write-backs skipped — and completes the future
135	* exceptionally with the original throwable.
136	*/
137	CompletableFuture<Void> execute() {
138	CompletableFuture<Void> visitsDone =
139	jobs.isEmpty() ? CompletableFuture.completedFuture(null) : runVisits();	1✔
140	CompletableFuture<Void> result = new CompletableFuture<>();	1✔
141	visitsDone.whenComplete(	1✔
142	(v, err) -> {
143	if (err != null) {	1✔
144	result.completeExceptionally(unwrap(err));	1✔
145	return;	1✔
146	}
147	try {
148	for (Runnable writeBack : writeBacks) {	1✔
149	writeBack.run();	1✔
150	}	1✔
151	result.complete(null);	1✔
152	} catch (Throwable t) {	×
153	result.completeExceptionally(t);	×
154	}	1✔
155	});	1✔
156	return result;	1✔
157	}
158
159	/** Run the visits with at most {@code concurrency} outstanding; fails with the first error. */
160	private CompletableFuture<Void> runVisits() {
161	AsyncSemaphore permits = new AsyncSemaphore(concurrency);	1✔
162	AtomicReference<Throwable> firstError = new AtomicReference<>();	1✔
163	List<CompletableFuture<Void>> all = new ArrayList<>(jobs.size());	1✔
164	for (LeafJob job : jobs) {	1✔
165	all.add(permits.acquire().thenCompose(p -> runJob(job, permits, firstError)));	1✔
166	}	1✔
167	return CompletableFuture.allOf(all.toArray(new CompletableFuture[0]))	1✔
168	.thenCompose(	1✔
169	v -> {
170	Throwable error = firstError.get();	1✔
171	if (error == null) {	1✔
172	return CompletableFuture.completedFuture(null);	1✔
173	}
174	CompletableFuture<Void> failed = new CompletableFuture<>();	1✔
175	failed.completeExceptionally(error);	1✔
176	return failed;	1✔
177	});
178	}
179
180	/** Runs one job under a held permit, releasing it exactly once when the visit settles. */
181	private CompletableFuture<Void> runJob(
182	LeafJob job, AsyncSemaphore permits, AtomicReference<Throwable> firstError) {
183	if (firstError.get() != null) {	1!
UNCOV 184	permits.release();	×
UNCOV 185	return CompletableFuture.completedFuture(null);	×
186	}
187	CompletableFuture<List<Payload>> visit;
188	try {
189	visit = payloadVisitor.visit(job.context, job.input);	1✔
190	} catch (RuntimeException \| Error t) {	1✔
191	firstError.compareAndSet(null, t);	1✔
192	permits.release();	1✔
193	return CompletableFuture.completedFuture(null);	1✔
194	}	1✔
195	if (visit == null) {	1!
196	firstError.compareAndSet(null, new IllegalStateException("payload visitor returned null"));	×
197	permits.release();	×
198	return CompletableFuture.completedFuture(null);	×
199	}
200	return visit	1✔
201	.handle(	1✔
202	(result, err) -> {
203	record(job, result, err, firstError);	1✔
204	return (Void) null;	1✔
205	})
206	.whenComplete((v, e) -> permits.release());	1✔
207	}
208
209	private void record(
210	LeafJob job, List<Payload> result, Throwable err, AtomicReference<Throwable> firstError) {
211	if (err != null) {	1✔
212	firstError.compareAndSet(null, unwrap(err));	1✔
213	} else if (result == null) {	1✔
214	firstError.compareAndSet(null, new IllegalStateException("payload visitor returned null"));	1✔
215	} else if (job.single && result.size() != 1) {	1✔
216	firstError.compareAndSet(	1✔
217	null,
218	new IllegalStateException(
219	"single-payload field requires exactly 1 returned payload, got " + result.size()));	1✔
220	} else {
221	job.result = result;	1✔
222	}
223	}	1✔
224
225	/** Strip the {@link CompletionException} a dependent stage wraps around its cause. */
226	private static Throwable unwrap(Throwable t) {
227	return (t instanceof CompletionException && t.getCause() != null) ? t.getCause() : t;	1!
228	}
229
230	/** A recorded visit and the slot its result lands in. */
231	private static final class LeafJob {
232	final List<Payload> input;
233	final Object context;
234	final boolean single;
235	volatile List<Payload> result;
236
237	LeafJob(List<Payload> input, Object context, boolean single) {	1✔
238	this.input = input;	1✔
239	this.context = context;	1✔
240	this.single = single;	1✔
241	}	1✔
242	}
243	}

temporalio / sdk-java / #347

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous