proposeTxn proposes a txn update, and then updates src to reflect the state of the commit after proposal is run.
(ctx context.Context, src *api.TxnContext)
| 304 | // proposeTxn proposes a txn update, and then updates src to reflect the state |
| 305 | // of the commit after proposal is run. |
| 306 | func (s *Server) proposeTxn(ctx context.Context, src *api.TxnContext) error { |
| 307 | var zp pb.ZeroProposal |
| 308 | zp.Txn = &api.TxnContext{ |
| 309 | StartTs: src.StartTs, |
| 310 | CommitTs: src.CommitTs, |
| 311 | Aborted: src.Aborted, |
| 312 | } |
| 313 | |
| 314 | // NOTE: It is important that we continue retrying proposeTxn until we succeed. This should |
| 315 | // happen, irrespective of what the user context timeout might be. We check for it before |
| 316 | // reaching this stage, but now that we're here, we have to ensure that the commit proposal goes |
| 317 | // through. Otherwise, we should block here forever. If we don't do this, we'll see txn |
| 318 | // violations in Jepsen, because we'll send out a MaxAssigned higher than a commit, which would |
| 319 | // cause newer txns to see older data. |
| 320 | |
| 321 | // If this node stops being the leader, we want this proposal to not be forwarded to the leader, |
| 322 | // and get aborted. |
| 323 | if err := s.Node.proposeAndWait(ctx, &zp); err != nil { |
| 324 | return err |
| 325 | } |
| 326 | |
| 327 | // There might be race between this proposal trying to commit and predicate |
| 328 | // move aborting it. A predicate move, triggered by Zero, would abort all |
| 329 | // pending transactions. At the same time, a client which has already done |
| 330 | // mutations, can proceed to commit it. A race condition can happen here, |
| 331 | // with both proposing their respective states, only one can succeed after |
| 332 | // the proposal is done. So, check again to see the fate of the transaction |
| 333 | // here. |
| 334 | src.CommitTs = s.orc.commitTs(src.StartTs) |
| 335 | if src.CommitTs == 0 { |
| 336 | src.Aborted = true |
| 337 | } |
| 338 | return nil |
| 339 | } |
| 340 | |
| 341 | func (s *Server) commit(ctx context.Context, src *api.TxnContext) error { |
| 342 | span := trace.SpanFromContext(ctx) |
no test coverage detected