30656653651

Committed 31 Jul 2026 06:48PM UTC coverage: 84.849% (+0.01%) from 84.839%

Build # 30656653651

Build Type

Pull #2137

gh-ci

Committed by

vchaindz

Commit Message

test(stdlib): expect the tx to survive a parse error

TestTx_Errors asserted the symptom that d491954a removes: a parse error used
to leave sqlTx nil, so IsClosed reported true, the session dropped the
transaction and the next statement answered "no transaction found". The
transaction now stays open, so that statement reaches the parser and returns
its own syntax error instead.

Assert that error, and roll back at the end - on the old code the rollback is
exactly what failed, so it pins the fix through the full client and session
path rather than only at the transactions package.

The sessions import goes with the ErrTransactionNotFound reference.

Pull Request Pull Request #2137: fix(server): keep the ongoing tx when a statement fails before execution

Coverage Stats

18 of 18 new or added lines in 2 files covered. (100.0%)

8 existing lines in 3 files now uncovered.

45282 of 53368 relevant lines covered (84.85%)

126521.27 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.65

/pkg/server/transaction.go

/*
Copyright 2026 Codenotary Inc. All rights reserved.

SPDX-License-Identifier: BUSL-1.1
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    https://mariadb.com/bsl11/

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package server

import (
        "context"
        "time"

        "github.com/codenotary/immudb/embedded/sql"
        "github.com/codenotary/immudb/pkg/api/schema"
        "github.com/codenotary/immudb/pkg/server/sessions"
        "github.com/golang/protobuf/ptypes/empty"
)

// BeginTx creates a new transaction. Only one read-write transaction per session can be active at a time.
func (s *ImmuServer) NewTx(ctx context.Context, request *schema.NewTxRequest) (*schema.NewTxResponse, error) {
        if request == nil {
                return nil, ErrIllegalArguments
        }
        if s.Options.GetMaintenance() {
                return nil, ErrNotAllowedInMaintenanceMode
        }

        if request.Mode == schema.TxMode_WriteOnly {
                // only in key-value mode, in sql we read catalog and write to it
                return nil, sessions.ErrWriteOnlyTXNotAllowed
        }

        sess, err := s.SessManager.GetSessionFromContext(ctx)
        if err != nil {
                return nil, err
        }

        opts := sql.DefaultTxOptions().
                WithReadOnly(request.Mode == schema.TxMode_ReadOnly)

        if request.SnapshotMustIncludeTxID != nil {
                opts.WithSnapshotMustIncludeTxID(func(_ uint64) uint64 {
                        return request.SnapshotMustIncludeTxID.GetValue()
                })
        }

        if request.SnapshotRenewalPeriod != nil {
                opts.WithSnapshotRenewalPeriod(time.Duration(request.SnapshotRenewalPeriod.GetValue()) * time.Millisecond)
        }

        opts.UnsafeMVCC = request.UnsafeMVCC

        tx, err := sess.NewTransaction(ctx, opts)
        if err != nil {
                return nil, err
        }

        return &schema.NewTxResponse{TransactionID: tx.GetID()}, nil
}

func (s *ImmuServer) Commit(ctx context.Context, _ *empty.Empty) (*schema.CommittedSQLTx, error) {
        if s.Options.GetMaintenance() {
                return nil, ErrNotAllowedInMaintenanceMode
        }

        tx, err := s.SessManager.GetTransactionFromContext(ctx)
        if err != nil {
                return nil, err
        }

        cTxs, err := s.SessManager.CommitTransaction(ctx, tx)
        if err != nil {
                return nil, err
        }

        cTx := cTxs[0]
        lastPKs := make(map[string]*schema.SQLValue, len(cTx.LastInsertedPKs()))
        for k, n := range cTx.LastInsertedPKs() {
                lastPKs[k] = &schema.SQLValue{Value: &schema.SQLValue_N{N: n}}
        }

        return &schema.CommittedSQLTx{
                Header:          schema.TxHeaderToProto(cTx.TxHeader()),
                UpdatedRows:     uint32(cTx.UpdatedRows()),
                LastInsertedPKs: lastPKs,
        }, nil
}

func (s *ImmuServer) Rollback(ctx context.Context, _ *empty.Empty) (*empty.Empty, error) {
        if s.Options.GetMaintenance() {
                return nil, ErrNotAllowedInMaintenanceMode
        }

        tx, err := s.SessManager.GetTransactionFromContext(ctx)
        if err != nil {
                return nil, err
        }

        return new(empty.Empty), s.SessManager.RollbackTransaction(tx)
}

func (s *ImmuServer) TxSQLExec(ctx context.Context, request *schema.SQLExecRequest) (*empty.Empty, error) {
        if request == nil {
                return nil, ErrIllegalArguments
        }

        if s.Options.GetMaintenance() {
                return new(empty.Empty), ErrNotAllowedInMaintenanceMode
        }

        tx, err := s.SessManager.GetTransactionFromContext(ctx)
        if err != nil {
                return new(empty.Empty), err
        }

        res := tx.SQLExec(ctx, request)

        if tx.IsClosed() {
                s.SessManager.DeleteTransaction(tx)
        }

        return new(empty.Empty), res
}

func (s *ImmuServer) TxSQLQuery(req *schema.SQLQueryRequest, srv schema.ImmuService_TxSQLQueryServer) error {
        if req == nil {
                return ErrIllegalArguments
        }
        if s.Options.GetMaintenance() {
                return ErrNotAllowedInMaintenanceMode
        }

        tx, err := s.SessManager.GetTransactionFromContext(srv.Context())
        if err != nil {
                return err
        }

        reader, err := tx.SQLQuery(srv.Context(), req)
        if err != nil {
                return err
        }
        defer reader.Close()

        return s.streamRows(srv.Context(), reader, tx.Database().MaxResultSize(), srv.Send)
}

1	/*
2	Copyright 2026 Codenotary Inc. All rights reserved.
3
4	SPDX-License-Identifier: BUSL-1.1
5	you may not use this file except in compliance with the License.
6	You may obtain a copy of the License at
7
8	https://mariadb.com/bsl11/
9
10	Unless required by applicable law or agreed to in writing, software
11	distributed under the License is distributed on an "AS IS" BASIS,
12	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	See the License for the specific language governing permissions and
14	limitations under the License.
15	*/
16
17	package server
18
19	import (
20	"context"
21	"time"
22
23	"github.com/codenotary/immudb/embedded/sql"
24	"github.com/codenotary/immudb/pkg/api/schema"
25	"github.com/codenotary/immudb/pkg/server/sessions"
26	"github.com/golang/protobuf/ptypes/empty"
27	)
28
29	// BeginTx creates a new transaction. Only one read-write transaction per session can be active at a time.
30	func (s ImmuServer) NewTx(ctx context.Context, request schema.NewTxRequest) (*schema.NewTxResponse, error) {	119✔
31	if request == nil {	120✔
32	return nil, ErrIllegalArguments	1✔
33	}	1✔
34	if s.Options.GetMaintenance() {	119✔
35	return nil, ErrNotAllowedInMaintenanceMode	1✔
36	}	1✔
37
38	if request.Mode == schema.TxMode_WriteOnly {	117✔
39	// only in key-value mode, in sql we read catalog and write to it	×
40	return nil, sessions.ErrWriteOnlyTXNotAllowed	×
41	}	×
42
43	sess, err := s.SessManager.GetSessionFromContext(ctx)	117✔
44	if err != nil {	117✔
45	return nil, err	×
46	}	×
47
48	opts := sql.DefaultTxOptions().	117✔
49	WithReadOnly(request.Mode == schema.TxMode_ReadOnly)	117✔
50		117✔
51	if request.SnapshotMustIncludeTxID != nil {	119✔
52	opts.WithSnapshotMustIncludeTxID(func(_ uint64) uint64 {	5✔
53	return request.SnapshotMustIncludeTxID.GetValue()	3✔
54	})	3✔
55	}
56
57	if request.SnapshotRenewalPeriod != nil {	119✔
58	opts.WithSnapshotRenewalPeriod(time.Duration(request.SnapshotRenewalPeriod.GetValue()) * time.Millisecond)	2✔
59	}	2✔
60
61	opts.UnsafeMVCC = request.UnsafeMVCC	117✔
62		117✔
63	tx, err := sess.NewTransaction(ctx, opts)	117✔
64	if err != nil {	117✔
65	return nil, err	×
66	}	×
67
68	return &schema.NewTxResponse{TransactionID: tx.GetID()}, nil	117✔
69	}
70
71	func (s ImmuServer) Commit(ctx context.Context, _ empty.Empty) (*schema.CommittedSQLTx, error) {	112✔
72	if s.Options.GetMaintenance() {	113✔
73	return nil, ErrNotAllowedInMaintenanceMode	1✔
74	}	1✔
75
76	tx, err := s.SessManager.GetTransactionFromContext(ctx)	111✔
77	if err != nil {	111✔
78	return nil, err	×
79	}	×
80
81	cTxs, err := s.SessManager.CommitTransaction(ctx, tx)	111✔
82	if err != nil {	113✔
83	return nil, err	2✔
84	}	2✔
85
86	cTx := cTxs[0]	109✔
87	lastPKs := make(map[string]*schema.SQLValue, len(cTx.LastInsertedPKs()))	109✔
88	for k, n := range cTx.LastInsertedPKs() {	109✔
89	lastPKs[k] = &schema.SQLValue{Value: &schema.SQLValue_N{N: n}}	×
90	}	×
91
92	return &schema.CommittedSQLTx{	109✔
93	Header: schema.TxHeaderToProto(cTx.TxHeader()),	109✔
94	UpdatedRows: uint32(cTx.UpdatedRows()),	109✔
95	LastInsertedPKs: lastPKs,	109✔
96	}, nil	109✔
97	}
98
99	func (s ImmuServer) Rollback(ctx context.Context, _ empty.Empty) (*empty.Empty, error) {	7✔
100	if s.Options.GetMaintenance() {	8✔
101	return nil, ErrNotAllowedInMaintenanceMode	1✔
102	}	1✔
103
104	tx, err := s.SessManager.GetTransactionFromContext(ctx)	6✔
105	if err != nil {	7✔
106	return nil, err	1✔
107	}	1✔
108
109	return new(empty.Empty), s.SessManager.RollbackTransaction(tx)	5✔
110	}
111
112	func (s ImmuServer) TxSQLExec(ctx context.Context, request schema.SQLExecRequest) (*empty.Empty, error) {	18✔
113	if request == nil {	19✔
114	return nil, ErrIllegalArguments	1✔
115	}	1✔
116
117	if s.Options.GetMaintenance() {	18✔
118	return new(empty.Empty), ErrNotAllowedInMaintenanceMode	1✔
119	}	1✔
120
121	tx, err := s.SessManager.GetTransactionFromContext(ctx)	16✔
122	if err != nil {	16✔
123	return new(empty.Empty), err	×
124	}	×
125
126	res := tx.SQLExec(ctx, request)	16✔
127		16✔
128	if tx.IsClosed() {	16✔
UNCOV 129	s.SessManager.DeleteTransaction(tx)	×
UNCOV 130	}	×
131
132	return new(empty.Empty), res	16✔
133	}
134
135	func (s ImmuServer) TxSQLQuery(req schema.SQLQueryRequest, srv schema.ImmuService_TxSQLQueryServer) error {	12✔
136	if req == nil {	13✔
137	return ErrIllegalArguments	1✔
138	}	1✔
139	if s.Options.GetMaintenance() {	12✔
140	return ErrNotAllowedInMaintenanceMode	1✔
141	}	1✔
142
143	tx, err := s.SessManager.GetTransactionFromContext(srv.Context())	10✔
144	if err != nil {	10✔
UNCOV 145	return err	×
UNCOV 146	}	×
147
148	reader, err := tx.SQLQuery(srv.Context(), req)	10✔
149	if err != nil {	12✔
150	return err	2✔
151	}	2✔
152	defer reader.Close()	8✔
153		8✔
154	return s.streamRows(srv.Context(), reader, tx.Database().MaxResultSize(), srv.Send)	8✔
155	}

codenotary / immudb / 30656653651

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous