Stripe's 2026 New Grad OA is the canonical "system-state simulation" problem: no algorithm trickery, but you must maintain 4-5 mutually-dependent state variables—per-server connection counts, connectId→server mapping, objId→server affinity binding, capacity caps, and re-routing after SHUTDOWN. The problem is split into 5 progressive parts, each layered on the previous code. This post lays out a full Python implementation across all 5 parts, with notes on the hidden-test traps.
Stripe NG OA Overview
| Dimension | Detail |
|---|---|
| Platform | HackerRank (full screen + camera recording) |
| Duration | 60 minutes |
| Questions | 1 problem, 5 parts |
| Difficulty | LeetCode Easy ~ Medium, logic-heavy |
| Pass threshold | All 5 hidden cases green |
| Languages | Python / Java / C++ |
Problem: Server Load Balancing System
Inputs:
numServer: total serversmaxConnection: per-server connection caprequests: a list of["CONNECT", connectId, userId, objId]/["DISCONNECT", connectId, userId, objId]/["SHUTDOWN", serverId]
Output: log of accepted operations.
The problem is to simulate the load balancer under the following progressive rules:
Part 1: Pure CONNECT Routing
Rule: route each CONNECT to the server with the smallest current connection count, breaking ties by lower index.
class LoadBalancer:
def __init__(self, num_server: int, max_connection: int):
self.num_server = num_server
self.max_connection = max_connection
self.counts = [0] * num_server
self.conn_to_server = {}
self.log = []
def _pick_server(self):
min_idx = 0
for i in range(1, self.num_server):
if self.counts[i] < self.counts[min_idx]:
min_idx = i
return min_idx
def connect(self, connect_id, user_id, obj_id):
s = self._pick_server()
self.counts[s] += 1
self.conn_to_server[connect_id] = s
self.log.append(("CONNECT", connect_id, s))
Part 1 key: maintain counts as a list, not a heap—later parts need indexing by objId/serverId, where heaps become awkward.
Part 2: Add DISCONNECT
Rule: release the server slot held by the given connectId.
def disconnect(self, connect_id, user_id, obj_id):
if connect_id not in self.conn_to_server:
return
s = self.conn_to_server.pop(connect_id)
self.counts[s] -= 1
self.log.append(("DISCONNECT", connect_id, s))
Hidden case: DISCONNECT for a non-existent connectId (never connected, or already cleared by SHUTDOWN) must silently no-op.
Part 3: objId Affinity Routing
Rule: connections sharing the same objId should go to the same server when possible.
def __init__(self, ...):
...
self.obj_to_server = {} # objId -> serverId
self.obj_count = defaultdict(int)
def connect(self, connect_id, user_id, obj_id):
if obj_id in self.obj_to_server:
s = self.obj_to_server[obj_id]
else:
s = self._pick_server()
self.obj_to_server[obj_id] = s
self.counts[s] += 1
self.obj_count[obj_id] += 1
self.conn_to_server[connect_id] = s
self.conn_to_obj[connect_id] = obj_id
self.log.append(("CONNECT", connect_id, s))
def disconnect(self, connect_id, user_id, obj_id):
if connect_id not in self.conn_to_server:
return
s = self.conn_to_server.pop(connect_id)
o = self.conn_to_obj.pop(connect_id)
self.counts[s] -= 1
self.obj_count[o] -= 1
if self.obj_count[o] == 0:
del self.obj_to_server[o]
self.log.append(("DISCONNECT", connect_id, s))
Part 3 key: when obj_count hits 0, remove the obj_to_server mapping—otherwise the same objId reconnecting later may re-route to a server that's been shut down (Part 5 fails otherwise).
Part 4: Capacity Cap
Rule: a server at maxConnection rejects new CONNECTs.
def connect(self, connect_id, user_id, obj_id):
if obj_id in self.obj_to_server:
s = self.obj_to_server[obj_id]
if self.counts[s] >= self.max_connection:
self.log.append(("REJECT", connect_id))
return
else:
s = self._pick_available_server()
if s is None:
self.log.append(("REJECT", connect_id))
return
self.obj_to_server[obj_id] = s
...
Part 4 design call: when the obj-bound server is full, do you fall back to another server? The expected answer is no—reject. This reflects Stripe's bias: "affinity priority > completion rate."
Part 5: SHUTDOWN Re-routing
Rule: on ["SHUTDOWN", serverId]:
- all connections on that server are invalidated (not counted as DISCONNECT)
- all objId bindings on that server are released
- subsequent CONNECTs bind to other servers
def shutdown(self, server_id):
if server_id in self.shut_down:
return
self.shut_down.add(server_id)
self.counts[server_id] = 0
for obj in [o for o, srv in self.obj_to_server.items() if srv == server_id]:
del self.obj_to_server[obj]
self.obj_count[obj] = 0
for cid in [c for c, srv in self.conn_to_server.items() if srv == server_id]:
del self.conn_to_server[cid]
if cid in self.conn_to_obj:
del self.conn_to_obj[cid]
self.log.append(("SHUTDOWN", server_id))
Three Part-5 hidden cases:
- all objId bindings on the shutdown server must be cleared (otherwise reconnecting that obj routes to a dead server)
- SHUTDOWN must NOT affect prior CONNECT log entries
- SHUTDOWN must be idempotent when applied to an already-shutdown server
Combined Solution (All 5 Parts)
from collections import defaultdict
from typing import List
class LoadBalancer:
def __init__(self, num_server: int, max_connection: int):
self.num_server = num_server
self.max_connection = max_connection
self.counts = [0] * num_server
self.shut_down = set()
self.conn_to_server = {}
self.conn_to_obj = {}
self.obj_to_server = {}
self.obj_count = defaultdict(int)
self.log = []
def _pick_available_server(self):
candidates = [
i for i in range(self.num_server)
if i not in self.shut_down and self.counts[i] < self.max_connection
]
if not candidates:
return None
return min(candidates, key=lambda i: (self.counts[i], i))
def connect(self, connect_id, user_id, obj_id):
if obj_id in self.obj_to_server:
s = self.obj_to_server[obj_id]
if s in self.shut_down or self.counts[s] >= self.max_connection:
self.log.append(("REJECT", connect_id))
return
else:
s = self._pick_available_server()
if s is None:
self.log.append(("REJECT", connect_id))
return
self.obj_to_server[obj_id] = s
self.counts[s] += 1
self.obj_count[obj_id] += 1
self.conn_to_server[connect_id] = s
self.conn_to_obj[connect_id] = obj_id
self.log.append(("CONNECT", connect_id, s))
def disconnect(self, connect_id, user_id, obj_id):
if connect_id not in self.conn_to_server:
return
s = self.conn_to_server.pop(connect_id)
o = self.conn_to_obj.pop(connect_id)
self.counts[s] -= 1
self.obj_count[o] -= 1
if self.obj_count[o] == 0 and o in self.obj_to_server:
del self.obj_to_server[o]
self.log.append(("DISCONNECT", connect_id, s))
def shutdown(self, server_id):
if server_id in self.shut_down:
return
self.shut_down.add(server_id)
self.counts[server_id] = 0
for obj in [o for o, srv in self.obj_to_server.items() if srv == server_id]:
del self.obj_to_server[obj]
self.obj_count[obj] = 0
for cid in [c for c, srv in self.conn_to_server.items() if srv == server_id]:
del self.conn_to_server[cid]
if cid in self.conn_to_obj:
del self.conn_to_obj[cid]
self.log.append(("SHUTDOWN", server_id))
def run(self, requests: List[List]):
for req in requests:
op = req[0]
if op == "CONNECT":
self.connect(req[1], req[2], req[3])
elif op == "DISCONNECT":
self.disconnect(req[1], req[2], req[3])
elif op == "SHUTDOWN":
self.shutdown(req[1])
return self.log
Time: O(n_server) per operation, O(R · n_server) overall (n_server ≤ 50)
Space: O(R + #objs)
4 Habits That Land 5/5
1) Hand-test after each part
Part 5's hidden cases are too dense to debug from rubric alone. After each part, build a 5-10-line request sequence and verify state transitions manually.
2) Initialize state in one place
The grader penalizes "scattered state." Declare every state field in __init__; don't dynamically attach attributes mid-run.
3) Be explicit about REJECTs
REJECTs may or may not need to appear in the main log. Some Stripe versions ask you to return the reject list separately. Re-read the prompt before submitting.
4) Idempotency
A shutdown server should be SHUTDOWN-able again silently; double-DISCONNECTing the same connectId should be a no-op.
FAQ
Does Stripe rotate the NG OA problem each year?
The surface changes, but the structure is stable: 5 progressive parts, multiple maps to maintain, last part is "reroute / undo." 2024-2026 all followed this template (brace expansion → shipping cost → server load balancing).
Is 60 minutes enough?
Yes—but invest the first 15 minutes in planning: list every state variable for Parts 1-5 up front. Diving straight in usually forces a refactor at Part 4-5.
Should I use a class?
Strongly yes. Five parts share state; a functional approach blows function signatures up to dozens of arguments. Stripe graders prefer OOP here.
Can I pass without finishing Part 5?
Yes. Parts 1-4 (~80% of tests green) typically clear Stripe's pass threshold. Part 5 is a stretch.
Where can I find more candidate reports?
1point3acres has fresh reports weekly. Search "Stripe NG OA 5 part," "Stripe server load balancing," or "Stripe 2026 OA."
Preparing for the Stripe NG OA?
oavoservice offers full real-time Stripe NG OA assistance: part decomposition, state-variable planning, end-to-end HackerRank support. We have complete breakdowns of every "5-part simulation" Stripe has used—server load balancing, subscription notifications, shipping cost, brace expansion—with optimized template code.
Add WeChat Coding0201 to book Stripe NG OA coaching.
#Stripe #StripeOA #NewGrad #LoadBalancing #Simulation #OARealQuestions
Contact
Email: [email protected]
Telegram: @OAVOProxy