You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by ya...@apache.org on 2026/10/12 06:19:09 UTC
(airavata-custos) branch fix-amie-packet-tx created (now 20e931b03)
This is an automated email from the ASF dual-hosted git repository.
yasithdev pushed a change to branch fix-amie-packet-tx
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
at 20e931b03 update membership status inside a transaction
This branch includes the following new commits:
new 20e931b03 update membership status inside a transaction
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(airavata-custos) 01/01: update membership status inside a transaction
Posted by ya...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yasithdev pushed a commit to branch fix-amie-packet-tx
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
commit 20e931b039d46aca73cf642f3f5d77a8ac26424c
Author: Yasith Jayawardana <ya...@icloud.com>
AuthorDate: Mon Oct 12 02:18:59 2026 -0400
update membership status inside a transaction
---
connectors/ACCESS/AMIE-Processor/handler/handler.go | 4 ++--
.../handler/request_account_inactivate.go | 2 +-
.../request_account_inactivate_integration_test.go | 6 +++---
.../handler/request_account_reactivate.go | 2 +-
.../handler/request_project_inactivate.go | 2 +-
.../handler/request_project_reactivate.go | 2 +-
pkg/service/compute_allocation_membership.go | 20 +++++++++++++-------
7 files changed, 22 insertions(+), 16 deletions(-)
diff --git a/connectors/ACCESS/AMIE-Processor/handler/handler.go b/connectors/ACCESS/AMIE-Processor/handler/handler.go
index 13e5c4ae5..a97867b4c 100644
--- a/connectors/ACCESS/AMIE-Processor/handler/handler.go
+++ b/connectors/ACCESS/AMIE-Processor/handler/handler.go
@@ -139,7 +139,7 @@ func ensureUserIdentity(ctx context.Context, svc *service.Service, userID, globa
// the given status. Returns the rows that were updated. Silently returns an
// empty slice when the project or user is unknown, the reply still goes back
// to AMIE but no state changes.
-func flipUserMemberships(ctx context.Context, svc *service.Service, projectID, userID string, status models.AllocationStatus) ([]models.ComputeAllocationMembership, error) {
+func flipUserMemberships(ctx context.Context, tx *sql.Tx, svc *service.Service, projectID, userID string, status models.AllocationStatus) ([]models.ComputeAllocationMembership, error) {
project, err := svc.GetProject(ctx, projectID)
if err != nil {
if errors.Is(err, service.ErrNotFound) {
@@ -161,7 +161,7 @@ func flipUserMemberships(ctx context.Context, svc *service.Service, projectID, u
if m.UserID != userID {
continue
}
- flipped, err := svc.UpdateMembershipStatus(ctx, m.ID, status)
+ flipped, err := svc.UpdateMembershipStatusTx(ctx, tx, m.ID, status)
if err != nil {
return nil, fmt.Errorf("update membership %s: %w", m.ID, err)
}
diff --git a/connectors/ACCESS/AMIE-Processor/handler/request_account_inactivate.go b/connectors/ACCESS/AMIE-Processor/handler/request_account_inactivate.go
index 4a0e58b4c..fc25d5540 100644
--- a/connectors/ACCESS/AMIE-Processor/handler/request_account_inactivate.go
+++ b/connectors/ACCESS/AMIE-Processor/handler/request_account_inactivate.go
@@ -65,7 +65,7 @@ func (h *RequestAccountInactivateHandler) Handle(ctx context.Context, tx *sql.Tx
return err
}
- flipped, err := flipUserMemberships(ctx, h.svc, projectID, personID, models.INACTIVE)
+ flipped, err := flipUserMemberships(ctx, tx, h.svc, projectID, personID, models.INACTIVE)
if err != nil {
return fmt.Errorf("request_account_inactivate: flip memberships: %w", err)
}
diff --git a/connectors/ACCESS/AMIE-Processor/handler/request_account_inactivate_integration_test.go b/connectors/ACCESS/AMIE-Processor/handler/request_account_inactivate_integration_test.go
index 85709bf9a..fd2e021bd 100644
--- a/connectors/ACCESS/AMIE-Processor/handler/request_account_inactivate_integration_test.go
+++ b/connectors/ACCESS/AMIE-Processor/handler/request_account_inactivate_integration_test.go
@@ -305,9 +305,6 @@ func TestRequestAccountInactivate_MissingRequiredField(t *testing.T) {
// TestRequestAccountInactivate_ReplyFailurePropagates asserts the handler
// surfaces a ReplyToPacket failure to the caller so the processor can decide
// to retry. The REPLY_SENT audit row must be absent when the reply failed.
-// (Note the broader atomicity gap: the membership flip is already committed
-// because UpdateMembershipStatus opens its own tx; this test only asserts
-// the error propagation + audit shape.)
func TestRequestAccountInactivate_ReplyFailurePropagates(t *testing.T) {
database := setupTestDB(t)
seed := seedMembership(t, database)
@@ -332,6 +329,9 @@ func TestRequestAccountInactivate_ReplyFailurePropagates(t *testing.T) {
if got := countAuditActions(t, database, pkt.ID, model.AuditReplySent); got != 0 {
t.Errorf("audit REPLY_SENT on failed reply: got %d, want 0", got)
}
+ if got := getMembershipStatus(t, database, seed.Membership.ID); got != string(models.ACTIVE) {
+ t.Errorf("membership after failed reply: got %q, want ACTIVE", got)
+ }
}
// silence unused-import false positives when only some tests reference these.
diff --git a/connectors/ACCESS/AMIE-Processor/handler/request_account_reactivate.go b/connectors/ACCESS/AMIE-Processor/handler/request_account_reactivate.go
index fc9e56e0a..e1353d4b6 100644
--- a/connectors/ACCESS/AMIE-Processor/handler/request_account_reactivate.go
+++ b/connectors/ACCESS/AMIE-Processor/handler/request_account_reactivate.go
@@ -65,7 +65,7 @@ func (h *RequestAccountReactivateHandler) Handle(ctx context.Context, tx *sql.Tx
return err
}
- flipped, err := flipUserMemberships(ctx, h.svc, projectID, personID, models.ACTIVE)
+ flipped, err := flipUserMemberships(ctx, tx, h.svc, projectID, personID, models.ACTIVE)
if err != nil {
return fmt.Errorf("request_account_reactivate: flip memberships: %w", err)
}
diff --git a/connectors/ACCESS/AMIE-Processor/handler/request_project_inactivate.go b/connectors/ACCESS/AMIE-Processor/handler/request_project_inactivate.go
index b14c1434d..0449b3768 100644
--- a/connectors/ACCESS/AMIE-Processor/handler/request_project_inactivate.go
+++ b/connectors/ACCESS/AMIE-Processor/handler/request_project_inactivate.go
@@ -114,7 +114,7 @@ func (h *RequestProjectInactivateHandler) deactivateAllocations(ctx context.Cont
return fmt.Errorf("list memberships for allocation %s: %w", a.ID, err)
}
for _, m := range members {
- if _, err := h.svc.UpdateMembershipStatus(ctx, m.ID, models.INACTIVE); err != nil {
+ if _, err := h.svc.UpdateMembershipStatusTx(ctx, tx, m.ID, models.INACTIVE); err != nil {
return fmt.Errorf("inactivate membership %s: %w", m.ID, err)
}
if err := h.auditSvc.Log(ctx, tx, packet.ID, eventID, model.AuditInactivateMembership, "compute_allocation_membership", m.ID,
diff --git a/connectors/ACCESS/AMIE-Processor/handler/request_project_reactivate.go b/connectors/ACCESS/AMIE-Processor/handler/request_project_reactivate.go
index 2ea169d8c..4af5e60a8 100644
--- a/connectors/ACCESS/AMIE-Processor/handler/request_project_reactivate.go
+++ b/connectors/ACCESS/AMIE-Processor/handler/request_project_reactivate.go
@@ -124,7 +124,7 @@ func (h *RequestProjectReactivateHandler) reactivateAllocationsAndPI(ctx context
if m.UserID != project.ProjectPIID {
continue
}
- if _, err := h.svc.UpdateMembershipStatus(ctx, m.ID, models.ACTIVE); err != nil {
+ if _, err := h.svc.UpdateMembershipStatusTx(ctx, tx, m.ID, models.ACTIVE); err != nil {
return fmt.Errorf("reactivate PI membership %s: %w", m.ID, err)
}
if err := h.auditSvc.Log(ctx, tx, packet.ID, eventID, model.AuditReactivateMembership, "compute_allocation_membership", m.ID,
diff --git a/pkg/service/compute_allocation_membership.go b/pkg/service/compute_allocation_membership.go
index 517679187..3ca4a677f 100644
--- a/pkg/service/compute_allocation_membership.go
+++ b/pkg/service/compute_allocation_membership.go
@@ -172,7 +172,15 @@ func (s *Service) UpdateComputeAllocationMembership(ctx context.Context, m *mode
// UpdateMembershipStatus sets the lifecycle status (ACTIVE, INACTIVE, etc.)
// of the membership identified by the given ID. Other fields are preserved.
-func (s *Service) UpdateMembershipStatus(ctx context.Context, id string, status models.AllocationStatus) (*models.ComputeAllocationMembership, error) {
+func (s *Service) UpdateMembershipStatus(ctx context.Context, id string, status models.AllocationStatus) (m *models.ComputeAllocationMembership, err error) {
+ err = s.inTx(ctx, func(tx *sql.Tx) (err error) {
+ m, err = s.UpdateMembershipStatusTx(ctx, tx, id, status)
+ return
+ })
+ return
+}
+
+func (s *Service) UpdateMembershipStatusTx(ctx context.Context, tx *sql.Tx, id string, status models.AllocationStatus) (*models.ComputeAllocationMembership, error) {
if id == "" {
return nil, fmt.Errorf("%w: compute allocation membership id is required", ErrInvalidInput)
}
@@ -187,12 +195,10 @@ func (s *Service) UpdateMembershipStatus(ctx context.Context, id string, status
return nil, ErrNotFound
}
existing.MembershipStatus = status
- if err := s.inTx(ctx, func(tx *sql.Tx) error {
- if err := s.memberships.Update(ctx, tx, existing); err != nil {
- return err
- }
- return s.eventBus.Publish(ctx, tx, events.ComputeAllocationMembershipUpdateEvent, existing)
- }); err != nil {
+ if err := s.memberships.Update(ctx, tx, existing); err != nil {
+ return nil, fmt.Errorf("update compute allocation membership status: %w", err)
+ }
+ if err := s.eventBus.Publish(ctx, tx, events.ComputeAllocationMembershipUpdateEvent, existing); err != nil {
return nil, fmt.Errorf("update compute allocation membership status: %w", err)
}
return existing, nil