Method serverTxn

client/v3/leasing/txn.go:191–223

()

}

func (txn *txnLeasing) serverTxn() (*v3.TxnResponse, error) {
	if err := txn.lkv.waitSession(txn.ctx); err != nil {
		return nil, err
	}

	userOps := gatherOps(append(txn.opst, txn.opse...))
	userTxn := v3.OpTxn(txn.cs, txn.opst, txn.opse)
	fbOps := txn.fallback(userOps)

	defer closeAll(txn.lkv.leases.LockWriteOps(userOps))
	for {
		cmps, err := txn.guard(userOps)
		if err != nil {
			return nil, err
		}
		resp, err := txn.lkv.kv.Txn(txn.ctx).If(cmps...).Then(userTxn).Else(fbOps...).Commit()
		if err != nil {
			for _, cmp := range cmps {
				txn.lkv.leases.Evict(strings.TrimPrefix(string(cmp.Key), txn.lkv.pfx))
			}
			return nil, err
		}
		if resp.Succeeded {
			txn.commitToCache((*v3pb.TxnResponse)(resp), userTxn)
			userResp := resp.Responses[0].GetResponseTxn()
			userResp.Header = resp.Header
			return (*v3.TxnResponse)(userResp), nil
		}
		if err := txn.revokeFallback(resp.Responses); err != nil {
			return nil, err
		}
	}
}