|
99 | 99 | import org.prebid.server.util.BidderUtil; |
100 | 100 | import org.prebid.server.util.HttpUtil; |
101 | 101 | import org.prebid.server.util.ListUtil; |
| 102 | +import org.prebid.server.util.MapUtil; |
102 | 103 | import org.prebid.server.util.PbsUtil; |
103 | 104 | import org.prebid.server.util.StreamUtil; |
104 | 105 |
|
|
115 | 116 | import java.util.Objects; |
116 | 117 | import java.util.Optional; |
117 | 118 | import java.util.Set; |
| 119 | +import java.util.function.Function; |
| 120 | +import java.util.function.Predicate; |
118 | 121 | import java.util.stream.Collectors; |
119 | 122 |
|
120 | 123 | public class ExchangeService { |
@@ -259,17 +262,7 @@ private Future<AuctionContext> runAuction(AuctionContext receivedContext) { |
259 | 262 | .map(receivedContext::with)) |
260 | 263 |
|
261 | 264 | .map(context -> updateRequestMetric(context, uidsCookie, aliases, account, requestTypeMetric)) |
262 | | - .compose(context -> Future.join( |
263 | | - context.getAuctionParticipations().stream() |
264 | | - .map(auctionParticipation -> processAndRequestBids( |
265 | | - context, |
266 | | - auctionParticipation.getBidderRequest(), |
267 | | - timeout, |
268 | | - aliases) |
269 | | - .map(auctionParticipation::with)) |
270 | | - .toList()) |
271 | | - // send all the requests to the bidders and gathers results |
272 | | - .map(CompositeFuture::<AuctionParticipation>list) |
| 265 | + .compose(context -> processAndRequestBids(context, timeout, aliases) |
273 | 266 | .map(storedResponseProcessor::updateStoredBidResponse) |
274 | 267 | .map(auctionParticipations -> storedResponseProcessor.mergeWithBidderResponses( |
275 | 268 | auctionParticipations, |
@@ -459,13 +452,10 @@ private void makeBidRejectionTrackers(Map<String, BidRejectionTracker> bidReject |
459 | 452 | bidderToImpIds.computeIfAbsent(bidder, bidderName -> new HashSet<>()).add(impId)); |
460 | 453 | } |
461 | 454 |
|
462 | | - bidderToImpIds.forEach((bidder, impIds) -> { |
463 | | - if (bidRejectionTrackers.containsKey(bidder)) { |
464 | | - bidRejectionTrackers.put(bidder, new BidRejectionTracker(bidRejectionTrackers.get(bidder), impIds)); |
465 | | - } else { |
466 | | - bidRejectionTrackers.put(bidder, new BidRejectionTracker(bidder, impIds, logSamplingRate)); |
467 | | - } |
468 | | - }); |
| 455 | + bidderToImpIds.forEach((bidder, impIds) -> bidRejectionTrackers.put(bidder, |
| 456 | + bidRejectionTrackers.containsKey(bidder) |
| 457 | + ? BidRejectionTracker.withAdditionalImpIds(bidRejectionTrackers.get(bidder), impIds) |
| 458 | + : new BidRejectionTracker(bidder, impIds, logSamplingRate))); |
469 | 459 | } |
470 | 460 |
|
471 | 461 | private static StoredResponseResult populateStoredResponse(StoredResponseResult storedResponseResult, |
@@ -1149,10 +1139,93 @@ private AuctionContext updateRequestMetric(AuctionContext context, |
1149 | 1139 | return context; |
1150 | 1140 | } |
1151 | 1141 |
|
1152 | | - private Future<BidderResponse> processAndRequestBids(AuctionContext auctionContext, |
1153 | | - BidderRequest bidderRequest, |
1154 | | - Timeout timeout, |
1155 | | - BidderAliases aliases) { |
| 1142 | + private Future<List<AuctionParticipation>> processAndRequestBids(AuctionContext auctionContext, |
| 1143 | + Timeout timeout, |
| 1144 | + BidderAliases aliases) { |
| 1145 | + |
| 1146 | + // when we ignore Futures from uncompleted secondary bidders, |
| 1147 | + // they continue running in the background and can write errors to BidRejectionTrackers. |
| 1148 | + // To prevent any issues, we provide them with a copy of BidRejectionTracker |
| 1149 | + // and only merge this copy back if secondary bidder has completed in time |
| 1150 | + final Map<String, BidRejectionTracker> copiedBidRejectionTrackers = |
| 1151 | + MapUtil.mapValues(auctionContext.getBidRejectionTrackers(), BidRejectionTracker::copyOf); |
| 1152 | + |
| 1153 | + final Map<String, AuctionParticipation> bidderToAuctionParticipation = auctionContext |
| 1154 | + .getAuctionParticipations().stream().collect(Collectors.toMap( |
| 1155 | + AuctionParticipation::getBidder, |
| 1156 | + Function.identity())); |
| 1157 | + |
| 1158 | + final Map<String, Future<BidderResponse>> bidderToFutureResponse = MapUtil.mapValues( |
| 1159 | + bidderToAuctionParticipation, |
| 1160 | + auctionParticipation -> processAndRequestBidsForSingleBidder( |
| 1161 | + auctionContext.with(copiedBidRejectionTrackers), |
| 1162 | + auctionParticipation.getBidderRequest(), |
| 1163 | + timeout, |
| 1164 | + aliases)); |
| 1165 | + |
| 1166 | + return buildPrimaryBiddersCompositeFuture(auctionContext, bidderToFutureResponse).transform(ignored -> { |
| 1167 | + mergeBidRejectionTrackers(auctionContext, copiedBidRejectionTrackers, bidderToFutureResponse); |
| 1168 | + |
| 1169 | + return Future.succeededFuture(bidderToFutureResponse.entrySet().stream() |
| 1170 | + .map(MapUtil.mapEntryValueMapper((bidder, futureResponse) -> |
| 1171 | + futureResponse.isComplete() ? futureResponse : recoverUncompletedSecondaryBidder(bidder))) |
| 1172 | + .map(MapUtil.mapEntryMapper((bidder, futureResponse) -> futureResponse.map(bidderResponse -> |
| 1173 | + bidderToAuctionParticipation.get(bidder).with(bidderResponse)))) |
| 1174 | + .map(Future::result) |
| 1175 | + .filter(Objects::nonNull) |
| 1176 | + .toList()); |
| 1177 | + }); |
| 1178 | + } |
| 1179 | + |
| 1180 | + private Future<BidderResponse> recoverUncompletedSecondaryBidder(String bidderName) { |
| 1181 | + final BidderSeatBid bidderSeatBid = BidderSeatBid.builder() |
| 1182 | + .warnings(Collections.singletonList( |
| 1183 | + BidderError.of("secondary bidder timed out, auction proceeded", BidderError.Type.timeout))) |
| 1184 | + .build(); |
| 1185 | + |
| 1186 | + return Future.succeededFuture(BidderResponse.of(bidderName, bidderSeatBid, 0)); |
| 1187 | + } |
| 1188 | + |
| 1189 | + private CompositeFuture buildPrimaryBiddersCompositeFuture( |
| 1190 | + AuctionContext auctionContext, |
| 1191 | + Map<String, Future<BidderResponse>> bidderToFutureResponse) { |
| 1192 | + |
| 1193 | + final Set<String> secondaryBidders = Optional.of(auctionContext) |
| 1194 | + .map(AuctionContext::getAccount) |
| 1195 | + .map(Account::getAuction) |
| 1196 | + .map(AccountAuctionConfig::getSecondaryBidders) |
| 1197 | + .orElse(Collections.emptySet()); |
| 1198 | + |
| 1199 | + final List<Future<BidderResponse>> primaryBiddersFutureResponses = bidderToFutureResponse.keySet().stream() |
| 1200 | + .filter(Predicate.not(secondaryBidders::contains)) |
| 1201 | + .map(bidderToFutureResponse::get) |
| 1202 | + .toList(); |
| 1203 | + |
| 1204 | + return Future.join(CollectionUtils.isNotEmpty(primaryBiddersFutureResponses) |
| 1205 | + ? primaryBiddersFutureResponses |
| 1206 | + : bidderToFutureResponse.values().stream().toList()); |
| 1207 | + } |
| 1208 | + |
| 1209 | + private void mergeBidRejectionTrackers( |
| 1210 | + AuctionContext auctionContext, |
| 1211 | + Map<String, BidRejectionTracker> newBidRejectionTrackers, |
| 1212 | + Map<String, Future<BidderResponse>> bidderToFutureResponse) { |
| 1213 | + |
| 1214 | + final Map<String, BidRejectionTracker> mergedBidRejectionTrackers = MapUtil.mapValues( |
| 1215 | + bidderToFutureResponse, |
| 1216 | + (bidder, futureResponse) -> futureResponse.isComplete() |
| 1217 | + ? newBidRejectionTrackers.get(bidder) |
| 1218 | + : auctionContext.getBidRejectionTrackers().get(bidder) |
| 1219 | + .rejectAll(BidRejectionReason.ERROR_TIMED_OUT)); |
| 1220 | + |
| 1221 | + auctionContext.getBidRejectionTrackers().clear(); |
| 1222 | + auctionContext.getBidRejectionTrackers().putAll(mergedBidRejectionTrackers); |
| 1223 | + } |
| 1224 | + |
| 1225 | + private Future<BidderResponse> processAndRequestBidsForSingleBidder(AuctionContext auctionContext, |
| 1226 | + BidderRequest bidderRequest, |
| 1227 | + Timeout timeout, |
| 1228 | + BidderAliases aliases) { |
1156 | 1229 |
|
1157 | 1230 | return bidderRequestPostProcessor.process(bidderRequest, aliases, auctionContext) |
1158 | 1231 | .compose(result -> invokeHooksAndRequestBids(auctionContext, result.getValue(), timeout, aliases) |
|
0 commit comments