Method readTx

xenomint/state.go:249–319

(
	ctx context.Context, req *types.Request)

}

func (s *State) readTx(
	ctx context.Context, req *types.Request) (ref *QueryTracker, resp *types.Response, err error,
) {
	var (
		id             = s.getSeq()
		ierr           error
		cnames, ctypes []string
		data           [][]interface{}
		querier        sqlQuerier
	)
	if s.level == sql.LevelReadUncommitted && atomic.LoadUint32(&s.hasSchemaChange) == 1 {
		// lock transaction
		s.Lock()
		defer s.Unlock()
		querier = s.handler
	} else {
		var tx *sql.Tx
		if tx, ierr = s.reader().Begin(); ierr != nil {
			err = errors.Wrap(ierr, "open tx failed")
			return
		}
		querier = tx
		defer func() {
			_ = tx.Rollback()
		}()
	}

	defer func() {
		if ctx.Err() != nil {
			log.WithError(ctx.Err()).WithFields(log.Fields{
				"req":       req,
				"id":        id,
				"dirtyRead": atomic.LoadUint32(&s.hasSchemaChange) != 1,
			}).Warning("read query canceled")
		}
	}()

	for i, v := range req.Payload.Queries {
		if cnames, ctypes, data, ierr = readSingle(ctx, querier, &v); ierr != nil {
			err = errors.Wrapf(ierr, "query at #%d failed", i)
			// Add to failed pool list
			s.Lock()
			s.pool.setFailed(req)
			s.Unlock()
			return
		}
	}
	// Build query response
	ref = &QueryTracker{Req: req}
	s.Lock()
	s.pool.enqueueRead(ref)
	s.Unlock()
	resp = &types.Response{
		Header: types.SignedResponseHeader{
			ResponseHeader: types.ResponseHeader{
				Request:     req.Header.RequestHeader,
				RequestHash: req.Header.Hash(),
				NodeID:      s.nodeID,