(req *commitReq)
| 127 | } |
| 128 | |
| 129 | func (r *Runtime) leaderDoCommit(req *commitReq) { |
| 130 | if req.log != nil { |
| 131 | // mis-use follower commit for leader |
| 132 | log.Fatal("INVALID EXISTING LOG FOR LEADER COMMIT") |
| 133 | return |
| 134 | } |
| 135 | |
| 136 | req.tm.Add("queue") |
| 137 | |
| 138 | // create leader log |
| 139 | var ( |
| 140 | l *kt.Log |
| 141 | logData []byte |
| 142 | cr = &commitResult{} |
| 143 | err error |
| 144 | ) |
| 145 | |
| 146 | logData = append(logData, r.uint64ToBytes(req.index)...) |
| 147 | logData = append(logData, r.uint64ToBytes(atomic.LoadUint64(&r.lastCommit))...) |
| 148 | |
| 149 | if l, err = r.newLog(req.ctx, kt.LogCommit, logData); err != nil { |
| 150 | // serve error, leader could not write log |
| 151 | return |
| 152 | } |
| 153 | |
| 154 | req.tm.Add("write_wal") |
| 155 | |
| 156 | // not wrapping underlying handler commit error |
| 157 | cr.result, err = r.doCommit(req.ctx, req.data, true) |
| 158 | |
| 159 | req.tm.Add("db_write") |
| 160 | |
| 161 | // mark last commit |
| 162 | atomic.StoreUint64(&r.lastCommit, l.Index) |
| 163 | |
| 164 | // send commit |
| 165 | cr.rpc = r.applyRPC(l, r.minCommitFollowers) |
| 166 | cr.index = l.Index |
| 167 | cr.err = err |
| 168 | |
| 169 | // TODO(): text log for rpc errors |
| 170 | |
| 171 | // TODO(): mark uncommitted nodes and remove from peers |
| 172 | |
| 173 | req.result.Set(cr) |
| 174 | |
| 175 | req.tm.Add("send_follower_commit") |
| 176 | |
| 177 | return |
| 178 | } |
| 179 | |
| 180 | func (r *Runtime) followerDoCommit(req *commitReq) { |
| 181 | if req.log == nil { |
no test coverage detected