1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package com.linecorp.centraldogma.internal.client;
17
18 import static com.google.common.base.Preconditions.checkArgument;
19 import static com.spotify.futures.CompletableFutures.exceptionallyCompletedFuture;
20 import static java.util.Objects.requireNonNull;
21 import static java.util.concurrent.CompletableFuture.completedFuture;
22
23 import java.util.LinkedHashMap;
24 import java.util.List;
25 import java.util.Map;
26 import java.util.Objects;
27 import java.util.Set;
28 import java.util.concurrent.CompletableFuture;
29 import java.util.concurrent.CompletionException;
30 import java.util.concurrent.ExecutionException;
31 import java.util.concurrent.ScheduledExecutorService;
32 import java.util.concurrent.TimeUnit;
33 import java.util.function.BiFunction;
34 import java.util.function.BiPredicate;
35 import java.util.function.Function;
36 import java.util.function.Supplier;
37
38 import org.jspecify.annotations.Nullable;
39 import org.slf4j.Logger;
40 import org.slf4j.LoggerFactory;
41
42 import com.google.common.annotations.VisibleForTesting;
43 import com.google.common.collect.ImmutableList;
44 import com.spotify.futures.CompletableFutures;
45
46 import com.linecorp.centraldogma.client.AbstractCentralDogma;
47 import com.linecorp.centraldogma.client.CentralDogma;
48 import com.linecorp.centraldogma.client.CentralDogmaRepository;
49 import com.linecorp.centraldogma.client.RepositoryInfo;
50 import com.linecorp.centraldogma.common.Author;
51 import com.linecorp.centraldogma.common.Change;
52 import com.linecorp.centraldogma.common.Commit;
53 import com.linecorp.centraldogma.common.Entry;
54 import com.linecorp.centraldogma.common.EntryType;
55 import com.linecorp.centraldogma.common.Markup;
56 import com.linecorp.centraldogma.common.MergeQuery;
57 import com.linecorp.centraldogma.common.MergedEntry;
58 import com.linecorp.centraldogma.common.PathPattern;
59 import com.linecorp.centraldogma.common.PushResult;
60 import com.linecorp.centraldogma.common.Query;
61 import com.linecorp.centraldogma.common.Revision;
62 import com.linecorp.centraldogma.common.RevisionNotFoundException;
63
64 import io.micrometer.core.instrument.MeterRegistry;
65
66
67
68
69
70 public final class ReplicationLagTolerantCentralDogma extends AbstractCentralDogma {
71
72 private static final Logger logger =
73 LoggerFactory.getLogger(ReplicationLagTolerantCentralDogma.class);
74
75 private final CentralDogma delegate;
76 private final int maxRetries;
77 private final long retryIntervalMillis;
78 private final Supplier<?> currentReplicaHintSupplier;
79
80 @VisibleForTesting
81 final Map<RepoId, Revision> latestKnownRevisions = new LinkedHashMap<RepoId, Revision>() {
82 private static final long serialVersionUID = 3587793379404809027L;
83
84 @Override
85 protected boolean removeEldestEntry(Map.Entry<RepoId, Revision> eldest) {
86
87 return size() > 8192;
88 }
89 };
90
91 public ReplicationLagTolerantCentralDogma(ScheduledExecutorService blockingTaskExecutor,
92 CentralDogma delegate, int maxRetries, long retryIntervalMillis,
93 Supplier<?> currentReplicaHintSupplier,
94 @Nullable MeterRegistry meterRegistry) {
95 super(blockingTaskExecutor, meterRegistry);
96
97 requireNonNull(delegate, "delegate");
98 checkArgument(maxRetries > 0, "maxRetries: %s (expected: > 0)", maxRetries);
99 checkArgument(retryIntervalMillis >= 0,
100 "retryIntervalMillis: %s (expected: >= 0)", retryIntervalMillis);
101 requireNonNull(currentReplicaHintSupplier, "currentReplicaHintSupplier");
102
103 this.delegate = delegate;
104 this.maxRetries = maxRetries;
105 this.retryIntervalMillis = retryIntervalMillis;
106 this.currentReplicaHintSupplier = currentReplicaHintSupplier;
107 }
108
109 @Override
110 public CentralDogma withAccessToken(String accessToken) {
111 requireNonNull(accessToken, "accessToken");
112 return new ReplicationLagTolerantCentralDogma(
113 executor(), delegate.withAccessToken(accessToken),
114 maxRetries, retryIntervalMillis, currentReplicaHintSupplier, meterRegistry());
115 }
116
117 @Override
118 public CompletableFuture<Void> createProject(String projectName) {
119 return delegate.createProject(projectName);
120 }
121
122 @Override
123 public CompletableFuture<Void> removeProject(String projectName) {
124 return delegate.removeProject(projectName).thenAccept(unused -> {
125 synchronized (latestKnownRevisions) {
126 latestKnownRevisions.entrySet().removeIf(e -> e.getKey().projectName.equals(projectName));
127 }
128 });
129 }
130
131 @Override
132 public CompletableFuture<Void> purgeProject(String projectName) {
133 return delegate.purgeProject(projectName);
134 }
135
136 @Override
137 public CompletableFuture<Void> unremoveProject(String projectName) {
138 return delegate.unremoveProject(projectName);
139 }
140
141 @Override
142 public CompletableFuture<Set<String>> listProjects() {
143 return delegate.listProjects();
144 }
145
146 @Override
147 public CompletableFuture<Set<String>> listRemovedProjects() {
148 return delegate.listRemovedProjects();
149 }
150
151 @Override
152 public CompletableFuture<CentralDogmaRepository> createRepository(String projectName,
153 String repositoryName) {
154 return delegate.createRepository(projectName, repositoryName);
155 }
156
157 @Override
158 public CompletableFuture<Void> removeRepository(String projectName, String repositoryName) {
159 return delegate.removeRepository(projectName, repositoryName).thenAccept(unused -> {
160 synchronized (latestKnownRevisions) {
161 latestKnownRevisions.remove(new RepoId(projectName, repositoryName));
162 }
163 });
164 }
165
166 @Override
167 public CompletableFuture<Void> purgeRepository(String projectName, String repositoryName) {
168 return delegate.purgeRepository(projectName, repositoryName);
169 }
170
171 @Override
172 public CompletableFuture<CentralDogmaRepository> unremoveRepository(String projectName,
173 String repositoryName) {
174 return delegate.unremoveRepository(projectName, repositoryName);
175 }
176
177 @Override
178 public CompletableFuture<Map<String, RepositoryInfo>> listRepositories(String projectName) {
179 return executeWithRetries(
180 new Supplier<CompletableFuture<Map<String, RepositoryInfo>>>() {
181 @Override
182 public CompletableFuture<Map<String, RepositoryInfo>> get() {
183 return delegate.listRepositories(projectName);
184 }
185
186 @Override
187 public String toString() {
188 return "listRepositories(" + projectName + ')';
189 }
190 },
191 (res, cause) -> {
192 if (res != null) {
193 for (RepositoryInfo info : res.values()) {
194 if (!updateLatestKnownRevision(projectName, info.name(), info.headRevision())) {
195 return true;
196 }
197 }
198 }
199 return false;
200 });
201 }
202
203 @Override
204 public CompletableFuture<Set<String>> listRemovedRepositories(String projectName) {
205 return delegate.listRemovedRepositories(projectName);
206 }
207
208 @Override
209 public CompletableFuture<Revision> normalizeRevision(
210 String projectName, String repositoryName, Revision revision) {
211 return executeWithRetries(
212 new Supplier<CompletableFuture<Revision>>() {
213 @Override
214 public CompletableFuture<Revision> get() {
215 return delegate.normalizeRevision(projectName, repositoryName, revision);
216 }
217
218 @Override
219 public String toString() {
220 return "normalizeRevision(" + projectName + ", " + repositoryName + ", " +
221 revision + ')';
222 }
223 },
224 (res, cause) -> {
225 if (cause != null) {
226 return handleRevisionNotFound(projectName, repositoryName, revision, cause);
227 }
228
229 if (revision.isRelative()) {
230 final Revision headRevision = res.forward(-(revision.major() + 1));
231 return !updateLatestKnownRevision(projectName, repositoryName, headRevision);
232 }
233
234 updateLatestKnownRevision(projectName, repositoryName, revision);
235 return false;
236 });
237 }
238
239 @Override
240 public CompletableFuture<Map<String, EntryType>> listFiles(
241 String projectName, String repositoryName, Revision revision, PathPattern pathPattern) {
242 return normalizeRevisionAndExecuteWithRetries(
243 projectName, repositoryName, revision,
244 new Function<Revision, CompletableFuture<Map<String, EntryType>>>() {
245 @Override
246 public CompletableFuture<Map<String, EntryType>> apply(Revision normRev) {
247 return delegate.listFiles(projectName, repositoryName, normRev, pathPattern);
248 }
249
250 @Override
251 public String toString() {
252 return "listFiles(" + projectName + ", " + repositoryName + ", " +
253 revision + ", " + pathPattern + ')';
254 }
255 });
256 }
257
258 @Override
259 public <T> CompletableFuture<Entry<T>> getFile(
260 String projectName, String repositoryName, Revision revision, Query<T> query,
261 boolean viewRaw, boolean renderTemplate, @Nullable String variableFile) {
262 return normalizeRevisionAndExecuteWithRetries(
263 projectName, repositoryName, revision,
264 new Function<>() {
265 @Override
266 public CompletableFuture<Entry<T>> apply(Revision normRev) {
267 return delegate.getFile(projectName, repositoryName, normRev, query,
268 viewRaw, renderTemplate, variableFile);
269 }
270
271 @Override
272 public String toString() {
273 return "getFile(" + projectName + ", " + repositoryName + ", " +
274 revision + ", " + query + ", " + viewRaw + ", " + renderTemplate + ", " +
275 variableFile + ')';
276 }
277 });
278 }
279
280 @Override
281 public CompletableFuture<Map<String, Entry<?>>> getFiles(
282 String projectName, String repositoryName, Revision revision, PathPattern pathPattern,
283 boolean viewRaw, boolean renderTemplate, @Nullable String variableFile) {
284 return normalizeRevisionAndExecuteWithRetries(
285 projectName, repositoryName, revision,
286 new Function<>() {
287 @Override
288 public CompletableFuture<Map<String, Entry<?>>> apply(Revision normRev) {
289 return delegate.getFiles(projectName, repositoryName, normRev, pathPattern,
290 viewRaw, renderTemplate, variableFile);
291 }
292
293 @Override
294 public String toString() {
295 return "getFiles(" + projectName + ", " + repositoryName + ", " +
296 revision + ", " + pathPattern + ", " + viewRaw + ", " + renderTemplate + ", " +
297 variableFile + ')';
298 }
299 });
300 }
301
302 @Override
303 public <T> CompletableFuture<MergedEntry<T>> mergeFiles(
304 String projectName, String repositoryName, Revision revision,
305 MergeQuery<T> mergeQuery) {
306 return normalizeRevisionAndExecuteWithRetries(
307 projectName, repositoryName, revision,
308 new Function<>() {
309 @Override
310 public CompletableFuture<MergedEntry<T>> apply(Revision normRev) {
311 return delegate.mergeFiles(projectName, repositoryName, normRev, mergeQuery);
312 }
313
314 @Override
315 public String toString() {
316 return "mergeFiles(" + projectName + ", " + repositoryName + ", " +
317 revision + ", " + mergeQuery + ')';
318 }
319 });
320 }
321
322 @Override
323 public CompletableFuture<List<Commit>> getHistory(
324 String projectName, String repositoryName, Revision from,
325 Revision to, PathPattern pathPattern, int maxCommits) {
326 return normalizeRevisionsAndExecuteWithRetries(
327 projectName, repositoryName, from, to,
328 new BiFunction<>() {
329 @Override
330 public CompletableFuture<List<Commit>> apply(Revision normFromRev, Revision normToRev) {
331 return delegate.getHistory(projectName, repositoryName,
332 normFromRev, normToRev, pathPattern, maxCommits);
333 }
334
335 @Override
336 public String toString() {
337 return "getHistory(" + projectName + ", " + repositoryName + ", " +
338 from + ", " + to + ", " + pathPattern + ", " + maxCommits + ')';
339 }
340 });
341 }
342
343 @Override
344 public <T> CompletableFuture<Change<T>> getDiff(
345 String projectName, String repositoryName, Revision from, Revision to, Query<T> query) {
346 return normalizeRevisionsAndExecuteWithRetries(
347 projectName, repositoryName, from, to,
348 new BiFunction<Revision, Revision, CompletableFuture<Change<T>>>() {
349 @Override
350 public CompletableFuture<Change<T>> apply(Revision normFromRev, Revision normToRev) {
351 return delegate.getDiff(projectName, repositoryName,
352 normFromRev, normToRev, query);
353 }
354
355 @Override
356 public String toString() {
357 return "getDiff(" + projectName + ", " + repositoryName + ", " +
358 from + ", " + to + ", " + query + ')';
359 }
360 });
361 }
362
363 @Override
364 public CompletableFuture<List<Change<?>>> getDiff(
365 String projectName, String repositoryName, Revision from, Revision to, PathPattern pathPattern) {
366 return normalizeRevisionsAndExecuteWithRetries(
367 projectName, repositoryName, from, to,
368 new BiFunction<Revision, Revision, CompletableFuture<List<Change<?>>>>() {
369 @Override
370 public CompletableFuture<List<Change<?>>> apply(Revision normFromRev, Revision normToRev) {
371 return delegate.getDiff(projectName, repositoryName,
372 normFromRev, normToRev, pathPattern);
373 }
374
375 @Override
376 public String toString() {
377 return "getDiffs(" + projectName + ", " + repositoryName + ", " +
378 from + ", " + to + ", " + pathPattern + ')';
379 }
380 });
381 }
382
383 @Override
384 public CompletableFuture<List<Change<?>>> getPreviewDiffs(
385 String projectName, String repositoryName, Revision baseRevision,
386 Iterable<? extends Change<?>> changes) {
387 return normalizeRevisionAndExecuteWithRetries(
388 projectName, repositoryName, baseRevision,
389 new Function<Revision, CompletableFuture<List<Change<?>>>>() {
390 @Override
391 public CompletableFuture<List<Change<?>>> apply(Revision normBaseRev) {
392 return delegate.getPreviewDiffs(projectName, repositoryName, normBaseRev, changes);
393 }
394
395 @Override
396 public String toString() {
397 return "getPreviewDiffs(" + projectName + ", " + repositoryName + ", " +
398 baseRevision + ", ...)";
399 }
400 });
401 }
402
403 @Override
404 public CompletableFuture<PushResult> push(
405 String projectName, String repositoryName, Revision baseRevision,
406 String summary, String detail, Markup markup, Iterable<? extends Change<?>> changes) {
407 return executeWithRetries(
408 new Supplier<CompletableFuture<PushResult>>() {
409 @Override
410 public CompletableFuture<PushResult> get() {
411 return delegate.push(projectName, repositoryName, baseRevision,
412 summary, detail, markup, changes);
413 }
414
415 @Override
416 public String toString() {
417 return "push(" + projectName + ", " + repositoryName + ", " +
418 baseRevision + ", " + summary + ", ...)";
419 }
420 },
421 pushRetryPredicate(projectName, repositoryName, baseRevision));
422 }
423
424 @Override
425 public CompletableFuture<PushResult> push(
426 String projectName, String repositoryName, Revision baseRevision,
427 Author author, String summary, String detail, Markup markup,
428 Iterable<? extends Change<?>> changes) {
429 return executeWithRetries(
430 new Supplier<CompletableFuture<PushResult>>() {
431 @Override
432 public CompletableFuture<PushResult> get() {
433 return delegate.push(projectName, repositoryName, baseRevision,
434 author, summary, detail, markup, changes);
435 }
436
437 @Override
438 public String toString() {
439 return "push(" + projectName + ", " + repositoryName + ", " +
440 baseRevision + ", " + summary + ", ...)";
441 }
442 },
443 pushRetryPredicate(projectName, repositoryName, baseRevision));
444 }
445
446 private BiPredicate<PushResult, Throwable> pushRetryPredicate(
447 String projectName, String repositoryName, Revision baseRevision) {
448
449 return (res, cause) -> {
450 if (cause != null) {
451 return handleRevisionNotFound(projectName, repositoryName, baseRevision, cause);
452 }
453
454 updateLatestKnownRevision(projectName, repositoryName, res.revision());
455 return false;
456 };
457 }
458
459 @Override
460 public CompletableFuture<Revision> watchRepository(
461 String projectName, String repositoryName, Revision lastKnownRevision,
462 PathPattern pathPattern, long timeoutMillis, boolean errorOnEntryNotFound) {
463
464 return normalizeRevisionAndExecuteWithRetries(
465 projectName, repositoryName, lastKnownRevision,
466 new Function<Revision, CompletableFuture<Revision>>() {
467 @Override
468 public CompletableFuture<Revision> apply(Revision normLastKnownRevision) {
469 return delegate.watchRepository(projectName, repositoryName, normLastKnownRevision,
470 pathPattern, timeoutMillis, errorOnEntryNotFound)
471 .thenApply(newLastKnownRevision -> {
472 if (newLastKnownRevision != null) {
473 updateLatestKnownRevision(projectName, repositoryName,
474 newLastKnownRevision);
475 }
476 return newLastKnownRevision;
477 });
478 }
479
480 @Override
481 public String toString() {
482 return "watchRepository(" + projectName + ", " + repositoryName + ", " +
483 lastKnownRevision + ", " + pathPattern + ", " + timeoutMillis + ", " +
484 errorOnEntryNotFound + ')';
485 }
486 });
487 }
488
489 @Override
490 public <T> CompletableFuture<Entry<T>> watchFile(
491 String projectName, String repositoryName, Revision lastKnownRevision,
492 Query<T> query, long timeoutMillis, boolean errorOnEntryNotFound,
493 boolean viewRaw, boolean renderTemplate, @Nullable String variableFile,
494 @Nullable Revision templateRevision) {
495
496 return normalizeRevisionAndExecuteWithRetries(
497 projectName, repositoryName, lastKnownRevision,
498 new Function<>() {
499 @Override
500 public CompletableFuture<Entry<T>> apply(Revision normLastKnownRevision) {
501 return delegate.watchFile(projectName, repositoryName, normLastKnownRevision,
502 query, timeoutMillis, errorOnEntryNotFound,
503 viewRaw, renderTemplate, variableFile, templateRevision)
504 .thenApply(entry -> {
505 if (entry != null) {
506 updateLatestKnownRevision(projectName, repositoryName,
507 entry.revision());
508 }
509 return entry;
510 });
511 }
512
513 @Override
514 public String toString() {
515 return "watchFile(" + projectName + ", " + repositoryName + ", " +
516 lastKnownRevision + ", " + query + ", " + timeoutMillis + ", " +
517 errorOnEntryNotFound + ", " + viewRaw + ", " + renderTemplate + ", " +
518 variableFile + ", " + templateRevision + ')';
519 }
520 });
521 }
522
523 @Override
524 public CompletableFuture<Void> whenEndpointReady() {
525 return delegate.whenEndpointReady();
526 }
527
528
529
530
531
532
533 private <T> CompletableFuture<T> normalizeRevisionAndExecuteWithRetries(
534 String projectName, String repositoryName, Revision revision,
535 Function<Revision, CompletableFuture<T>> taskRunner) {
536 return normalizeRevision(projectName, repositoryName, revision)
537 .thenCompose(normRev -> executeWithRetries(
538 new Supplier<>() {
539 @Override
540 public CompletableFuture<T> get() {
541 return taskRunner.apply(normRev);
542 }
543
544 @Override
545 public String toString() {
546 return taskRunner + " with " + normRev;
547 }
548 },
549 (res, cause) -> cause != null &&
550 handleRevisionNotFound(projectName, repositoryName, normRev, cause)));
551 }
552
553
554
555
556
557
558
559 private <T> CompletableFuture<T> normalizeRevisionsAndExecuteWithRetries(
560 String projectName, String repositoryName, Revision from, Revision to,
561 BiFunction<Revision, Revision, CompletableFuture<T>> taskRunner) {
562
563 if (to == null) {
564 return exceptionallyCompletedFuture(new NullPointerException("to"));
565 }
566
567 if (from == null) {
568 return exceptionallyCompletedFuture(new NullPointerException("from"));
569 }
570
571 if (from.isRelative() && to.isRelative() ||
572 !from.isRelative() && !to.isRelative()) {
573
574
575
576 final int distance = to.major() - from.major();
577 final Revision baseRevision = to.compareTo(from) >= 0 ? to : from;
578
579 return normalizeRevision(projectName, repositoryName, baseRevision).thenCompose(normBaseRev -> {
580 final Revision normFromRev;
581 final Revision normToRev;
582 if (distance >= 0) {
583 normToRev = normBaseRev;
584 normFromRev = normBaseRev.backward(distance);
585 } else {
586 normFromRev = normBaseRev;
587 normToRev = normBaseRev.backward(-distance);
588 }
589
590 return executeWithRetries(
591 new Supplier<CompletableFuture<T>>() {
592 @Override
593 public CompletableFuture<T> get() {
594 return taskRunner.apply(normFromRev, normToRev);
595 }
596
597 @Override
598 public String toString() {
599 return taskRunner + " with [" + normFromRev + ", " + normToRev + ']';
600 }
601 },
602 (res, cause) -> {
603 if (cause == null) {
604 return false;
605 }
606 return handleRevisionNotFound(projectName, repositoryName, normBaseRev, cause);
607 });
608 });
609 } else {
610
611
612 return CompletableFutures.allAsList(ImmutableList.of(
613 normalizeRevision(projectName, repositoryName, from),
614 normalizeRevision(projectName, repositoryName, to))).thenCompose(normRevs -> {
615 final Revision normFromRev = normRevs.get(0);
616 final Revision normToRev = normRevs.get(1);
617 return executeWithRetries(
618 new Supplier<CompletableFuture<T>>() {
619 @Override
620 public CompletableFuture<T> get() {
621 return taskRunner.apply(normFromRev, normToRev);
622 }
623
624 @Override
625 public String toString() {
626 return taskRunner + " with [" + normFromRev + ", " + normToRev + ']';
627 }
628 },
629 (res, cause) -> {
630 if (cause == null) {
631 return false;
632 }
633
634 final Revision normBaseRev = normFromRev.compareTo(normToRev) > 0 ? normFromRev
635 : normToRev;
636 return handleRevisionNotFound(projectName, repositoryName, normBaseRev, cause);
637 });
638 });
639 }
640 }
641
642
643
644
645
646
647 private <T> CompletableFuture<T> executeWithRetries(
648 Supplier<CompletableFuture<T>> taskRunner,
649 BiPredicate<T, Throwable> retryPredicate) {
650 return executeWithRetries(taskRunner, retryPredicate, 0);
651 }
652
653 private <T> CompletableFuture<T> executeWithRetries(
654 Supplier<CompletableFuture<T>> taskRunner,
655 BiPredicate<T, Throwable> retryPredicate,
656 int attemptsSoFar) {
657
658 return CompletableFutures.handleCompose(taskRunner.get(), (res, cause) -> {
659 final Object currentReplicaHint = currentReplicaHintSupplier.get();
660 final int nextAttemptsSoFar = attemptsSoFar + 1;
661 final boolean retryRequired = retryPredicate.test(res, cause);
662 if (!retryRequired || nextAttemptsSoFar > maxRetries) {
663 if (retryRequired) {
664 if (currentReplicaHint != null) {
665 logger.warn("[{}] Failed to retrieve the up-to-date data from Central Dogma " +
666 "after {} retries: {} => {}",
667 currentReplicaHint, attemptsSoFar, taskRunner, resultOrCause(res, cause));
668 } else {
669 logger.warn("Failed to retrieve the up-to-date data from Central Dogma " +
670 "after {} retries: {} => {}",
671 attemptsSoFar, taskRunner, resultOrCause(res, cause));
672 }
673 } else if (logger.isDebugEnabled()) {
674 if (currentReplicaHint != null) {
675 logger.debug("[{}] Retrieved the up-to-date data after {} retries: {} => {}",
676 currentReplicaHint, attemptsSoFar, taskRunner, resultOrCause(res, cause));
677 } else {
678 logger.debug("Retrieved the up-to-date data after {} retries: {} => {}",
679 attemptsSoFar, taskRunner, resultOrCause(res, cause));
680 }
681 }
682
683 if (cause == null) {
684 return completedFuture(res);
685 } else {
686 return exceptionallyCompletedFuture(cause);
687 }
688 }
689
690 if (logger.isDebugEnabled()) {
691 if (currentReplicaHint != null) {
692 logger.debug("[{}] Got the out-of-date data ({} attempt(s) so far): {} => {}",
693 currentReplicaHint, nextAttemptsSoFar, taskRunner, resultOrCause(res, cause));
694 } else {
695 logger.debug("Got the out-of-date data ({} attempt(s) so far): {} => {}",
696 nextAttemptsSoFar, taskRunner, resultOrCause(res, cause));
697 }
698 }
699
700 final CompletableFuture<T> nextAttemptFuture = new CompletableFuture<>();
701 executor().schedule(() -> {
702 try {
703 executeWithRetries(taskRunner, retryPredicate,
704 nextAttemptsSoFar).handle((newRes, newCause) -> {
705 if (newCause != null) {
706 nextAttemptFuture.completeExceptionally(newCause);
707 } else {
708 nextAttemptFuture.complete(newRes);
709 }
710 return null;
711 });
712 } catch (Throwable t) {
713 nextAttemptFuture.completeExceptionally(t);
714 }
715 }, retryIntervalMillis, TimeUnit.MILLISECONDS);
716
717 return nextAttemptFuture;
718 }).toCompletableFuture();
719 }
720
721
722
723
724
725 private static Throwable peel(Throwable throwable) {
726 Throwable cause = throwable.getCause();
727 while (cause != null && cause != throwable &&
728 (throwable instanceof CompletionException || throwable instanceof ExecutionException)) {
729 throwable = cause;
730 cause = throwable.getCause();
731 }
732 return throwable;
733 }
734
735
736
737
738
739 private boolean handleRevisionNotFound(
740 String projectName, String repositoryName, Revision revision, Throwable cause) {
741 requireNonNull(cause, "cause");
742 cause = peel(cause);
743 if (!(cause instanceof RevisionNotFoundException)) {
744 return false;
745 }
746
747 final Revision latestKnownRevision = latestKnownRevision(projectName, repositoryName);
748 if (latestKnownRevision == null) {
749 return false;
750 }
751
752 if (revision.isRelative()) {
753 return revision.major() + latestKnownRevision.major() >= 0;
754 } else {
755 return revision.major() <= latestKnownRevision.major();
756 }
757 }
758
759 @Nullable
760 @VisibleForTesting
761 Revision latestKnownRevision(String projectName, String repositoryName) {
762 synchronized (latestKnownRevisions) {
763 return latestKnownRevisions.get(new RepoId(projectName, repositoryName));
764 }
765 }
766
767
768
769
770
771
772
773 private boolean updateLatestKnownRevision(String projectName, String repositoryName, Revision newRevision) {
774 final Object currentReplicaHint = currentReplicaHintSupplier.get();
775 final RepoId id = new RepoId(projectName, repositoryName);
776 synchronized (latestKnownRevisions) {
777 final Revision oldRevision = latestKnownRevisions.get(id);
778 if (oldRevision == null) {
779 if (currentReplicaHint != null) {
780 logger.debug("[{}] Updating the latest known revision for {}/{} from <unknown> to: {}",
781 currentReplicaHint, projectName, repositoryName, newRevision);
782 } else {
783 logger.debug("Updating the latest known revision for {}/{} from <unknown> to: {}",
784 projectName, repositoryName, newRevision);
785 }
786 latestKnownRevisions.put(id, newRevision);
787 return true;
788 }
789
790 final int comparison = oldRevision.compareTo(newRevision);
791 if (comparison < 0) {
792 if (currentReplicaHint != null) {
793 logger.debug("[{}] Updating the latest known revision for {}/{} from {} to: {}",
794 currentReplicaHint, projectName, repositoryName, oldRevision, newRevision);
795 } else {
796 logger.debug("Updating the latest known revision for {}/{} from {} to: {}",
797 projectName, repositoryName, oldRevision, newRevision);
798 }
799 latestKnownRevisions.put(id, newRevision);
800 return true;
801 }
802
803 if (comparison == 0) {
804 if (currentReplicaHint != null) {
805 logger.debug("[{}] The latest known revision for {}/{} stays unchanged at: {}",
806 currentReplicaHint, projectName, repositoryName, newRevision);
807 } else {
808 logger.debug("The latest known revision for {}/{} stays unchanged at: {}",
809 projectName, repositoryName, newRevision);
810 }
811 return true;
812 }
813
814 if (currentReplicaHint != null) {
815 logger.debug("[{}] An out-of-date latest known revision for {}/{}: {}",
816 currentReplicaHint, projectName, repositoryName, newRevision);
817 } else {
818 logger.debug("An out-of-date latest known revision for {}/{}: {}",
819 projectName, repositoryName, newRevision);
820 }
821 return false;
822 }
823 }
824
825 @Nullable
826 private static Object resultOrCause(@Nullable Object res, @Nullable Throwable cause) {
827 return res != null ? res : cause;
828 }
829
830 @Override
831 public void close() throws Exception {
832 delegate.close();
833 }
834
835 @VisibleForTesting
836 static final class RepoId {
837 final String projectName;
838 final String repositoryName;
839
840 RepoId(String projectName, String repositoryName) {
841 this.projectName = projectName;
842 this.repositoryName = repositoryName;
843 }
844
845 @Override
846 public boolean equals(Object o) {
847 if (this == o) {
848 return true;
849 }
850 if (!(o instanceof RepoId)) {
851 return false;
852 }
853 final RepoId that = (RepoId) o;
854 return projectName.equals(that.projectName) && repositoryName.equals(that.repositoryName);
855 }
856
857 @Override
858 public int hashCode() {
859 return Objects.hash(projectName, repositoryName);
860 }
861
862 @Override
863 public String toString() {
864 return projectName + '/' + repositoryName;
865 }
866 }
867 }