1924576561

Committed 11 Jul 2025 12:09PM UTC coverage: 65.521% (+0.01%) from 65.508%

Build # 1924576561

Build Type

push

gitlab-ci

Committed by

web-flow

Commit Message

Merge pull request #781 from mendersoftware/lock-smith

Created interface and implementation of distributed locks

Run Details

34 of 42 new or added lines in 1 file covered. (80.95%)

32381 of 49421 relevant lines covered (65.52%)

1.38 hits per line

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

80.95

/backend/pkg/mongo/lock.go

package mongo

import (
        "context"
        "fmt"
        "time"

        "go.mongodb.org/mongo-driver/bson"
        "go.mongodb.org/mongo-driver/mongo"
        "go.mongodb.org/mongo-driver/mongo/options"
        "go.mongodb.org/mongo-driver/mongo/writeconcern"

        "github.com/mendersoftware/mender-server/pkg/sync"
)

type distributedLock struct {
        coll   *mongo.Collection
        lockID string
        lock   lockDoc
}

func NewLockGenerator(db *mongo.Database, collName string) sync.DistributedLockGenerator {
        return func(lockID string) (sync.DistributedLock, error) {
                return NewLock(db, collName, lockID), nil
        }
}

// NewLock creates a DistributedLock using a single document in a collection.
// The implementation relies on the uniqueness of the primary index (_id), and
// uses a combination of timestamps with document versioning to prevent
// "stealing" locks aquired by other processes.
func NewLock(db *mongo.Database, collName, key string) sync.DistributedLock {
        coll := db.Collection(collName, options.
                Collection().
                SetWriteConcern(writeconcern.Majority()))
        return &distributedLock{
                coll:   coll,
                lockID: key,
        }
}

// lockDoc is the document describing the lock.
type lockDoc struct {
        // ID is the unique (resource) ID of the lock
        ID string `bson:"_id"`
        // Exp is the lock expiry time
        Exp time.Time `bson:"exp"`
        // Created is the time the lock was created (aquired)
        Created time.Time `bson:"created"`
        // Version is an attribute that is always incremented when aquiring the lock
        // this ensures (together with exp) that once lock is aquired it will not
        // release another process' if it expires.
        Version uint32 `bson:"v"`
}

func (l *distributedLock) TryLock(ctx context.Context) (bool, error) {
        const defaultTTL = time.Minute * 2
        deadline, ok := ctx.Deadline()
        if !ok {
                deadline = time.Now().Add(defaultTTL)
        }
        now := time.Now()
        var lock lockDoc
        err := l.coll.FindOneAndUpdate(ctx,
                bson.M{"_id": l.lockID, "exp": bson.M{"$lt": now}},
                bson.M{
                        "$set": bson.M{"_id": l.lockID, "exp": deadline, "created": now},
                        "$inc": bson.M{"v": 1},
                },
                options.FindOneAndUpdate().
                        SetReturnDocument(options.After).
                        SetUpsert(true)).
                Decode(&lock)
        if mongo.IsDuplicateKeyError(err) {
                return false, nil
        } else if err != nil {
                return false, fmt.Errorf("failed to aquire lock: %w", err)
        }
        l.lock = lock
        return true, nil
}

func (l *distributedLock) Unlock(ctx context.Context) error {
        _, err := l.coll.DeleteOne(ctx, l.lock)
        return err
}

1	package mongo
2
3	import (
4	"context"
5	"fmt"
6	"time"
7
8	"go.mongodb.org/mongo-driver/bson"
9	"go.mongodb.org/mongo-driver/mongo"
10	"go.mongodb.org/mongo-driver/mongo/options"
11	"go.mongodb.org/mongo-driver/mongo/writeconcern"
12
13	"github.com/mendersoftware/mender-server/pkg/sync"
14	)
15
16	type distributedLock struct {
17	coll *mongo.Collection
18	lockID string
19	lock lockDoc
20	}
21
NEW 22	func NewLockGenerator(db *mongo.Database, collName string) sync.DistributedLockGenerator {	×
NEW 23	return func(lockID string) (sync.DistributedLock, error) {	×
NEW 24	return NewLock(db, collName, lockID), nil	×
NEW 25	}	×
26	}
27
28	// NewLock creates a DistributedLock using a single document in a collection.
29	// The implementation relies on the uniqueness of the primary index (_id), and
30	// uses a combination of timestamps with document versioning to prevent
31	// "stealing" locks aquired by other processes.
32	func NewLock(db *mongo.Database, collName, key string) sync.DistributedLock {	1✔
33	coll := db.Collection(collName, options.	1✔
34	Collection().	1✔
35	SetWriteConcern(writeconcern.Majority()))	1✔
36	return &distributedLock{	1✔
37	coll: coll,	1✔
38	lockID: key,	1✔
39	}	1✔
40	}	1✔
41
42	// lockDoc is the document describing the lock.
43	type lockDoc struct {
44	// ID is the unique (resource) ID of the lock
45	ID string `bson:"_id"`
46	// Exp is the lock expiry time
47	Exp time.Time `bson:"exp"`
48	// Created is the time the lock was created (aquired)
49	Created time.Time `bson:"created"`
50	// Version is an attribute that is always incremented when aquiring the lock
51	// this ensures (together with exp) that once lock is aquired it will not
52	// release another process' if it expires.
53	Version uint32 `bson:"v"`
54	}
55
56	func (l *distributedLock) TryLock(ctx context.Context) (bool, error) {	1✔
57	const defaultTTL = time.Minute * 2	1✔
58	deadline, ok := ctx.Deadline()	1✔
59	if !ok {	1✔
NEW 60	deadline = time.Now().Add(defaultTTL)	×
NEW 61	}	×
62	now := time.Now()	1✔
63	var lock lockDoc	1✔
64	err := l.coll.FindOneAndUpdate(ctx,	1✔
65	bson.M{"_id": l.lockID, "exp": bson.M{"$lt": now}},	1✔
66	bson.M{	1✔
67	"$set": bson.M{"_id": l.lockID, "exp": deadline, "created": now},	1✔
68	"$inc": bson.M{"v": 1},	1✔
69	},	1✔
70	options.FindOneAndUpdate().	1✔
71	SetReturnDocument(options.After).	1✔
72	SetUpsert(true)).	1✔
73	Decode(&lock)	1✔
74	if mongo.IsDuplicateKeyError(err) {	2✔
75	return false, nil	1✔
76	} else if err != nil {	2✔
NEW 77	return false, fmt.Errorf("failed to aquire lock: %w", err)	×
NEW 78	}	×
79	l.lock = lock	1✔
80	return true, nil	1✔
81	}
82
83	func (l *distributedLock) Unlock(ctx context.Context) error {	1✔
84	_, err := l.coll.DeleteOne(ctx, l.lock)	1✔
85	return err	1✔
86	}	1✔

mendersoftware / mender-server / 1924576561

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