Skip to content

Commit 873cca9

Browse files
MariemBaccariMariemBaccari
committed
[raft/rid] Add rid Raftstore ISAs
1 parent 06c4bad commit 873cca9

5 files changed

Lines changed: 468 additions & 23 deletions

File tree

‎pkg/models/models.go‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package models
22

33
import (
4+
"encoding/json"
45
"strconv"
56
"time"
67

@@ -175,3 +176,26 @@ func (v *Version) ToTimestamp() *time.Time {
175176
}
176177
return &v.t
177178
}
179+
180+
func (v *Version) MarshalJSON() ([]byte, error) {
181+
return json.Marshal(v.String())
182+
}
183+
184+
func (v *Version) UnmarshalJSON(data []byte) error {
185+
var s string
186+
if err := json.Unmarshal(data, &s); err != nil {
187+
return err
188+
}
189+
190+
if s == "" {
191+
return nil
192+
}
193+
194+
parsed, err := VersionFromString(s)
195+
if err != nil {
196+
return stacktrace.Propagate(err, "failed to unmarshal version")
197+
}
198+
199+
*v = *parsed
200+
return nil
201+
}

‎pkg/rid/application/isa.go‎

Lines changed: 40 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import (
1010
dssmodels "github.com/interuss/dss/pkg/models"
1111
ridmodels "github.com/interuss/dss/pkg/rid/models"
1212
"github.com/interuss/dss/pkg/rid/repos"
13+
ridraftstore "github.com/interuss/dss/pkg/rid/store/raftstore"
1314
"github.com/interuss/stacktrace"
1415
)
1516

@@ -62,8 +63,13 @@ func (a *app) DeleteISA(ctx context.Context, id dssmodels.ID, owner dssmodels.Ow
6263
ret *ridmodels.IdentificationServiceArea
6364
subs []*ridmodels.Subscription
6465
)
66+
6567
// The following will automatically retry TXN retry errors.
66-
_, err := a.store.Transact(ctx, "", nil, func(ctx context.Context, repo repos.Repository) error {
68+
raftResult, err := a.store.Transact(ctx, ridraftstore.DeleteISATransaction, ridraftstore.DeleteISATransactionPayload{
69+
ID: id,
70+
Owner: owner,
71+
Version: version,
72+
}, func(ctx context.Context, repo repos.Repository) error {
6773
old, err := repo.GetISA(ctx, id, true)
6874
switch {
6975
case err != nil:
@@ -89,6 +95,20 @@ func (a *app) DeleteISA(ctx context.Context, id dssmodels.ID, owner dssmodels.Ow
8995
}
9096
return nil
9197
})
98+
99+
if err == nil && raftResult != nil {
100+
if result, ok := raftResult.(*ridraftstore.ISATransactionResult); ok {
101+
if result.Ret != nil {
102+
ret = result.Ret
103+
}
104+
if result.Subs != nil {
105+
subs = result.Subs
106+
}
107+
} else {
108+
return nil, nil, stacktrace.NewError("invalid result type: %T", raftResult)
109+
}
110+
}
111+
92112
return ret, subs, err // No need to Propagate this error as this stack layer does not add useful information
93113
}
94114

@@ -104,7 +124,7 @@ func (a *app) InsertISA(ctx context.Context, isa *ridmodels.IdentificationServic
104124
subs []*ridmodels.Subscription
105125
)
106126
// The following will automatically retry TXN retry errors.
107-
_, err := a.store.Transact(ctx, "", nil, func(ctx context.Context, repo repos.Repository) error {
127+
raftResult, err := a.store.Transact(ctx, ridraftstore.InsertISATransaction, isa, func(ctx context.Context, repo repos.Repository) error {
108128
// ensure it doesn't exist yet
109129
old, err := repo.GetISA(ctx, isa.ID, false)
110130
if err != nil {
@@ -127,6 +147,14 @@ func (a *app) InsertISA(ctx context.Context, isa *ridmodels.IdentificationServic
127147
}
128148
return nil
129149
})
150+
if err == nil && raftResult != nil {
151+
if result, ok := raftResult.(*ridraftstore.ISATransactionResult); ok {
152+
ret = result.Ret
153+
subs = result.Subs
154+
} else {
155+
return nil, nil, stacktrace.NewError("invalid result type: %T", raftResult)
156+
}
157+
}
130158
return ret, subs, err // No need to Propagate this error as this stack layer does not add useful information
131159
}
132160

@@ -138,7 +166,7 @@ func (a *app) UpdateISA(ctx context.Context, isa *ridmodels.IdentificationServic
138166
subs []*ridmodels.Subscription
139167
)
140168
// The following will automatically retry TXN retry errors.
141-
_, err := a.store.Transact(ctx, "", nil, func(ctx context.Context, repo repos.Repository) error {
169+
raftResult, err := a.store.Transact(ctx, ridraftstore.UpdateISATransaction, isa, func(ctx context.Context, repo repos.Repository) error {
142170
var err error
143171

144172
old, err := repo.GetISA(ctx, isa.ID, true)
@@ -178,5 +206,14 @@ func (a *app) UpdateISA(ctx context.Context, isa *ridmodels.IdentificationServic
178206
return nil
179207
})
180208

209+
if err == nil && raftResult != nil {
210+
if result, ok := raftResult.(*ridraftstore.ISATransactionResult); ok {
211+
ret = result.Ret
212+
subs = result.Subs
213+
} else {
214+
return nil, nil, stacktrace.NewError("invalid result type: %T", raftResult)
215+
}
216+
}
217+
181218
return ret, subs, err // No need to Propagate this error as this stack layer does not add useful information
182219
}

‎pkg/rid/store/raftstore/identification_service_area.go‎

Lines changed: 128 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -5,36 +5,149 @@ import (
55
"time"
66

77
"github.com/golang/geo/s2"
8-
dsserr "github.com/interuss/dss/pkg/errors"
98
dssmodels "github.com/interuss/dss/pkg/models"
109
ridmodels "github.com/interuss/dss/pkg/rid/models"
1110
"github.com/interuss/stacktrace"
1211
)
1312

14-
func (r *repo) GetISA(_ context.Context, id dssmodels.ID, forUpdate bool) (*ridmodels.IdentificationServiceArea, error) {
15-
return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "GetISA not implemented for raftstore")
13+
type getISAPayload struct {
14+
ID dssmodels.ID
15+
ForUpdate bool
1616
}
1717

18-
func (r *repo) DeleteISA(_ context.Context, isa *ridmodels.IdentificationServiceArea) (*ridmodels.IdentificationServiceArea, error) {
19-
return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "DeleteISA not implemented for raftstore")
18+
func (r *repo) GetISA(ctx context.Context, id dssmodels.ID, forUpdate bool) (*ridmodels.IdentificationServiceArea, error) {
19+
result, err := r.consensus.ProposeValue(ctx, storeID, getISA, &getISAPayload{ID: id, ForUpdate: forUpdate}, true)
20+
if err != nil {
21+
return nil, err
22+
}
23+
24+
if result == nil {
25+
return nil, nil
26+
}
27+
28+
isa, ok := result.(*ridmodels.IdentificationServiceArea)
29+
if !ok {
30+
return nil, stacktrace.NewError("invalid result type: %T", result)
31+
}
32+
33+
return isa, nil
2034
}
2135

22-
func (r *repo) InsertISA(_ context.Context, isa *ridmodels.IdentificationServiceArea) (*ridmodels.IdentificationServiceArea, error) {
23-
return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "InsertISA not implemented for raftstore")
36+
func (r *repo) DeleteISA(ctx context.Context, isa *ridmodels.IdentificationServiceArea) (*ridmodels.IdentificationServiceArea, error) {
37+
result, err := r.consensus.ProposeValue(ctx, storeID, deleteISA, isa, false)
38+
if err != nil {
39+
return nil, err
40+
}
41+
42+
if result == nil {
43+
return nil, nil
44+
}
45+
46+
isa, ok := result.(*ridmodels.IdentificationServiceArea)
47+
if !ok {
48+
return nil, stacktrace.NewError("invalid result type: %T", result)
49+
}
50+
51+
return isa, nil
2452
}
2553

26-
func (r *repo) UpdateISA(_ context.Context, isa *ridmodels.IdentificationServiceArea) (*ridmodels.IdentificationServiceArea, error) {
27-
return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "UpdateISA not implemented for raftstore")
54+
func (r *repo) InsertISA(ctx context.Context, isa *ridmodels.IdentificationServiceArea) (*ridmodels.IdentificationServiceArea, error) {
55+
result, err := r.consensus.ProposeValue(ctx, storeID, insertISA, isa, false)
56+
if err != nil {
57+
return nil, err
58+
}
59+
60+
if result == nil {
61+
return nil, nil
62+
}
63+
64+
isa, ok := result.(*ridmodels.IdentificationServiceArea)
65+
if !ok {
66+
return nil, stacktrace.NewError("invalid result type: %T", result)
67+
}
68+
69+
return isa, nil
2870
}
2971

30-
func (r *repo) SearchISAs(_ context.Context, cells s2.CellUnion, earliest *time.Time, latest *time.Time) ([]*ridmodels.IdentificationServiceArea, error) {
31-
return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "SearchISAs not implemented for raftstore")
72+
func (r *repo) UpdateISA(ctx context.Context, isa *ridmodels.IdentificationServiceArea) (*ridmodels.IdentificationServiceArea, error) {
73+
result, err := r.consensus.ProposeValue(ctx, storeID, updateISA, isa, false)
74+
if err != nil {
75+
return nil, err
76+
}
77+
78+
if result == nil {
79+
return nil, nil
80+
}
81+
82+
isa, ok := result.(*ridmodels.IdentificationServiceArea)
83+
if !ok {
84+
return nil, stacktrace.NewError("invalid result type: %T", result)
85+
}
86+
87+
return isa, nil
3288
}
3389

34-
func (r *repo) ListExpiredISAs(_ context.Context, writer string, threshold time.Time) ([]*ridmodels.IdentificationServiceArea, error) {
35-
return nil, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "ListExpiredISAs not implemented for raftstore")
90+
type searchISAsPayload struct {
91+
Cells s2.CellUnion
92+
Earliest *time.Time
93+
Latest *time.Time
3694
}
3795

38-
func (r *repo) CountISAs(_ context.Context) (int64, error) {
39-
return 0, stacktrace.NewErrorWithCode(dsserr.NotImplemented, "CountISAs not implemented for raftstore")
96+
func (r *repo) SearchISAs(ctx context.Context, cells s2.CellUnion, earliest *time.Time, latest *time.Time) ([]*ridmodels.IdentificationServiceArea, error) {
97+
result, err := r.consensus.ProposeValue(ctx, storeID, searchISAs, &searchISAsPayload{Cells: cells, Earliest: earliest, Latest: latest}, true)
98+
if err != nil {
99+
return nil, err
100+
}
101+
102+
if result == nil {
103+
return nil, nil
104+
}
105+
106+
isa, ok := result.([]*ridmodels.IdentificationServiceArea)
107+
if !ok {
108+
return nil, stacktrace.NewError("invalid result type: %T", result)
109+
}
110+
111+
return isa, nil
112+
}
113+
114+
type listExpiredISAsPayload struct {
115+
Writer string
116+
Threshold time.Time
117+
}
118+
119+
func (r *repo) ListExpiredISAs(ctx context.Context, writer string, threshold time.Time) ([]*ridmodels.IdentificationServiceArea, error) {
120+
result, err := r.consensus.ProposeValue(ctx, storeID, listExpiredISAs, &listExpiredISAsPayload{Writer: writer, Threshold: threshold}, true)
121+
if err != nil {
122+
return nil, err
123+
}
124+
125+
if result == nil {
126+
return nil, nil
127+
}
128+
129+
isa, ok := result.([]*ridmodels.IdentificationServiceArea)
130+
if !ok {
131+
return nil, stacktrace.NewError("invalid result type: %T", result)
132+
}
133+
134+
return isa, nil
135+
}
136+
137+
func (r *repo) CountISAs(ctx context.Context) (int64, error) {
138+
result, err := r.consensus.ProposeValue(ctx, storeID, countISAs, nil, true)
139+
if err != nil {
140+
return 0, err
141+
}
142+
143+
if result == nil {
144+
return 0, nil
145+
}
146+
147+
count, ok := result.(int64)
148+
if !ok {
149+
return 0, stacktrace.NewError("invalid result type: %T", result)
150+
}
151+
152+
return count, nil
40153
}

0 commit comments

Comments
 (0)