Skip to content

refactor(api): move RI exchange and Marketplace provider wiring into one composition file #1565

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
IAmJSD merged 2 commits into from
Oct 28, 2026
Merged

refactor(api): move RI exchange and Marketplace provider wiring into one composition file #1565

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5cb2cb0
refactor(api): move RI exchange and Marketplace provider wiring into …
IAmJSD Oct 27, 2026
bb548fe
Merge remote-tracking branch 'origin/main' into fix/issue-41-ri-excha…
IAmJSD Oct 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 5 additions & 7 deletions internal/api/handler_marketplace.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,6 @@ import (
"time"

"github.com/LeanerCloud/cloud-commitments-go/pkg/logging"
awsprovider "github.com/LeanerCloud/cloud-commitments-go/providers/aws"
ec2svc "github.com/LeanerCloud/cloud-commitments-go/providers/aws/services/ec2"
"github.com/LeanerCloud/cloud-commitments-platform/internal/auth"
"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
"github.com/LeanerCloud/cloud-commitments-platform/internal/credentials"
Expand All @@ -38,9 +36,9 @@ import (
// marketplaceEC2Client is the narrow EC2 interface the marketplace handlers
// need. Using a minimal interface keeps test stubs small.
type marketplaceEC2Client interface {
CreateMarketplaceListing(ctx context.Context, req ec2svc.MarketplaceListingRequest) (ec2svc.MarketplaceListingResult, error)
DescribeMarketplaceListing(ctx context.Context, listingID string) (ec2svc.MarketplaceListingResult, error)
CancelMarketplaceListing(ctx context.Context, listingID string) (ec2svc.MarketplaceListingResult, error)
CreateMarketplaceListing(ctx context.Context, req ec2MarketplaceListingRequest) (ec2MarketplaceListingResult, error)
DescribeMarketplaceListing(ctx context.Context, listingID string) (ec2MarketplaceListingResult, error)
CancelMarketplaceListing(ctx context.Context, listingID string) (ec2MarketplaceListingResult, error)
// FetchOfferingClass calls AWS DescribeReservedInstances to determine
// whether a given Reserved Instance is 'standard' or 'convertible'. Used
// when the purchase_history row has no offering_class stored (pre-migration
Expand All @@ -54,7 +52,7 @@ func (h *Handler) buildMarketplaceEC2Client(cfg aws.Config) marketplaceEC2Client
if h.marketplaceEC2Factory != nil {
return h.marketplaceEC2Factory(cfg)
}
return awsprovider.NewEC2ClientDirect(cfg)
return newMarketplaceEC2Client(cfg)
}

// loadMarketplaceAWSConfig returns an AWS config for the purchase's region that
Expand Down Expand Up @@ -304,7 +302,7 @@ const marketplaceCompensationTimeout = 10 * time.Second

// keepUncanceledListing records a listing that could be neither persisted nor
// canceled; releasing the claim instead would hide it from the cancel endpoint.
func (h *Handler) keepUncanceledListing(ctx context.Context, purchaseID string, listing ec2svc.MarketplaceListingResult, cancelErr error) error {
func (h *Handler) keepUncanceledListing(ctx context.Context, purchaseID string, listing ec2MarketplaceListingResult, cancelErr error) error {
logging.Errorf("marketplace: rollback cancel for listing %s failed: %v", listing.ListingID, cancelErr)
state := listing.State
if state == "" {
Expand Down
55 changes: 27 additions & 28 deletions internal/api/handler_marketplace_attempt.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@ import (
"strings"

"github.com/LeanerCloud/cloud-commitments-go/pkg/logging"
ec2svc "github.com/LeanerCloud/cloud-commitments-go/providers/aws/services/ec2"
"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
"github.com/aws/aws-sdk-go-v2/aws"
sdkec2 "github.com/aws/aws-sdk-go-v2/service/ec2"
Expand All @@ -29,7 +28,7 @@ import (
// poller reads a listing's sale details through it too (issue #27), because
// the wrapper's DescribeMarketplaceListing returns only the state.
type marketplaceListingFinder interface {
FindMarketplaceListingByToken(ctx context.Context, reservedInstancesID, clientToken string) (ec2svc.MarketplaceListingResult, bool, error)
FindMarketplaceListingByToken(ctx context.Context, reservedInstancesID, clientToken string) (ec2MarketplaceListingResult, bool, error)
DescribeMarketplaceListingStatus(ctx context.Context, listingID string) (marketplaceListingStatus, error)
}

Expand All @@ -39,20 +38,20 @@ type describeListingsAPI interface {

type awsMarketplaceListingFinder struct{ api describeListingsAPI }

func (f awsMarketplaceListingFinder) FindMarketplaceListingByToken(ctx context.Context, reservedInstancesID, clientToken string) (ec2svc.MarketplaceListingResult, bool, error) {
func (f awsMarketplaceListingFinder) FindMarketplaceListingByToken(ctx context.Context, reservedInstancesID, clientToken string) (listing ec2MarketplaceListingResult, found bool, err error) {
out, err := f.api.DescribeReservedInstancesListings(ctx, &sdkec2.DescribeReservedInstancesListingsInput{
ReservedInstancesId: aws.String(reservedInstancesID),
})
if err != nil {
return ec2svc.MarketplaceListingResult{}, false, fmt.Errorf("DescribeReservedInstancesListings failed: %w", err)
return ec2MarketplaceListingResult{}, false, fmt.Errorf("DescribeReservedInstancesListings failed: %w", err)
}
for i := range out.ReservedInstancesListings {
l := &out.ReservedInstancesListings[i]
if aws.ToString(l.ClientToken) == clientToken {
return ec2svc.MarketplaceListingResult{ListingID: aws.ToString(l.ReservedInstancesListingId), State: string(l.Status)}, true, nil
return ec2MarketplaceListingResult{ListingID: aws.ToString(l.ReservedInstancesListingId), State: string(l.Status)}, true, nil
}
}
return ec2svc.MarketplaceListingResult{}, false, nil
return ec2MarketplaceListingResult{}, false, nil
}

func (h *Handler) buildMarketplaceFinder(cfg aws.Config) marketplaceListingFinder {
Expand All @@ -78,10 +77,10 @@ func marketplaceInstanceCount(row *config.PurchaseHistoryRecord) (int32, error)
return 1, nil
}

func toAWSPriceSchedule(schedule []MarketplacePriceTier) []ec2svc.MarketplacePriceTier {
out := make([]ec2svc.MarketplacePriceTier, 0, len(schedule))
func toAWSPriceSchedule(schedule []MarketplacePriceTier) []ec2MarketplacePriceTier {
out := make([]ec2MarketplacePriceTier, 0, len(schedule))
for _, t := range schedule {
out = append(out, ec2svc.MarketplacePriceTier{Term: t.TermMonths, Price: t.Price})
out = append(out, ec2MarketplacePriceTier{Term: t.TermMonths, Price: t.Price})
}
return out
}
Expand All @@ -93,14 +92,14 @@ func toAWSPriceSchedule(schedule []MarketplacePriceTier) []ec2svc.MarketplacePri
// does not leave the row stuck in the transient pending state. Returns the
// persisted listing and the schedule it was created with, which on a retry is
// the stored one, not the one passed in.
func (h *Handler) reserveAndCreateListing(ctx context.Context, purchaseID string, row *config.PurchaseHistoryRecord, cfg aws.Config, ec2Client marketplaceEC2Client, schedule []MarketplacePriceTier) (ec2svc.MarketplaceListingResult, []MarketplacePriceTier, error) {
func (h *Handler) reserveAndCreateListing(ctx context.Context, purchaseID string, row *config.PurchaseHistoryRecord, cfg aws.Config, ec2Client marketplaceEC2Client, schedule []MarketplacePriceTier) (ec2MarketplaceListingResult, []MarketplacePriceTier, error) {
instanceCount, err := marketplaceInstanceCount(row)
if err != nil {
return ec2svc.MarketplaceListingResult{}, nil, err
return ec2MarketplaceListingResult{}, nil, err
}
claim, used, err := h.claimListingSlot(ctx, purchaseID, schedule)
if err != nil {
return ec2svc.MarketplaceListingResult{}, nil, err
return ec2MarketplaceListingResult{}, nil, err
}
if claim.Resumed {
result, found, recErr := h.reconcileAttempt(ctx, purchaseID, claim, h.buildMarketplaceFinder(cfg))
Expand Down Expand Up @@ -147,28 +146,28 @@ func (h *Handler) claimListingSlot(ctx context.Context, purchaseID string, sched
// resolved: result holds the recorded listing, or err why it ended in a dead
// one. Not found means AWS has nothing for the token, so the caller creates
// the listing with that same token.
func (h *Handler) reconcileAttempt(ctx context.Context, purchaseID string, claim *config.MarketplaceListingClaim, finder marketplaceListingFinder) (ec2svc.MarketplaceListingResult, bool, error) {
listing, found, err := finder.FindMarketplaceListingByToken(ctx, purchaseID, claim.ClientToken)
func (h *Handler) reconcileAttempt(ctx context.Context, purchaseID string, claim *config.MarketplaceListingClaim, finder marketplaceListingFinder) (listing ec2MarketplaceListingResult, found bool, err error) {
listing, found, err = finder.FindMarketplaceListingByToken(ctx, purchaseID, claim.ClientToken)
if err != nil {
h.releaseMarketplaceClaim(ctx, purchaseID, claim, true)
return ec2svc.MarketplaceListingResult{}, false, mapAWSMarketplaceError("could not reconcile the earlier listing attempt on AWS", err)
return ec2MarketplaceListingResult{}, false, mapAWSMarketplaceError("could not reconcile the earlier listing attempt on AWS", err)
}
if !found {
return ec2svc.MarketplaceListingResult{}, false, nil
return ec2MarketplaceListingResult{}, false, nil
}
if listing.State == "" {
listing.State = config.ListingStatePending
}
logging.Warnf("marketplace: purchase %s: earlier attempt already created listing %s (%s); recording it", purchaseID, listing.ListingID, listing.State)
if isDeadListingState(listing.State) {
return ec2svc.MarketplaceListingResult{}, true, h.recordDeadListing(ctx, purchaseID, claim, listing)
return ec2MarketplaceListingResult{}, true, h.recordDeadListing(ctx, purchaseID, claim, listing)
}
if err = h.config.UpdatePurchaseHistoryListing(ctx, purchaseID, listing.ListingID, listing.State); err != nil {
if isListingEndedConflict(err) {
return ec2svc.MarketplaceListingResult{}, true, listingCanceledConcurrently(listing.ListingID)
return ec2MarketplaceListingResult{}, true, listingCanceledConcurrently(listing.ListingID)
}
h.releaseMarketplaceClaim(ctx, purchaseID, claim, true)
return ec2svc.MarketplaceListingResult{}, true, fmt.Errorf("listing %s exists on AWS but could not be recorded: %w", listing.ListingID, err)
return ec2MarketplaceListingResult{}, true, fmt.Errorf("listing %s exists on AWS but could not be recorded: %w", listing.ListingID, err)
}
return listing, true, nil
}
Expand All @@ -184,7 +183,7 @@ func listingCanceledConcurrently(listingID string) error {
// "not listed" and let the next request create a second listing, so the
// listing is read back, and recorded as pending (live, cancelable) if its
// state is still unknown.
func createdListingState(ctx context.Context, ec2Client marketplaceEC2Client, created ec2svc.MarketplaceListingResult) string {
func createdListingState(ctx context.Context, ec2Client marketplaceEC2Client, created ec2MarketplaceListingResult) string {
if created.State != "" {
return created.State
}
Expand All @@ -196,8 +195,8 @@ func createdListingState(ctx context.Context, ec2Client marketplaceEC2Client, cr
return config.ListingStatePending
}

func (h *Handler) createAndPersistListing(ctx context.Context, purchaseID string, claim *config.MarketplaceListingClaim, ec2Client marketplaceEC2Client, instanceCount int32, awsSchedule []ec2svc.MarketplacePriceTier) (ec2svc.MarketplaceListingResult, error) {
result, err := ec2Client.CreateMarketplaceListing(ctx, ec2svc.MarketplaceListingRequest{
func (h *Handler) createAndPersistListing(ctx context.Context, purchaseID string, claim *config.MarketplaceListingClaim, ec2Client marketplaceEC2Client, instanceCount int32, awsSchedule []ec2MarketplacePriceTier) (ec2MarketplaceListingResult, error) {
result, err := ec2Client.CreateMarketplaceListing(ctx, ec2MarketplaceListingRequest{
ReservedInstancesID: purchaseID,
ClientToken: claim.ClientToken,
PriceSchedule: awsSchedule,
Expand All @@ -210,12 +209,12 @@ func (h *Handler) createAndPersistListing(ctx context.Context, purchaseID string
// then is the token dropped; otherwise the retry resends it.
h.releaseMarketplaceClaim(ctx, purchaseID, claim, !isClientFaultError(mapped))
logging.Warnf("marketplace: CreateReservedInstancesListing for purchase %s failed: %v", purchaseID, err)
return ec2svc.MarketplaceListingResult{}, mapped
return ec2MarketplaceListingResult{}, mapped
}

result.State = createdListingState(ctx, ec2Client, result)
if isDeadListingState(result.State) {
return ec2svc.MarketplaceListingResult{}, h.recordDeadListing(ctx, purchaseID, claim, result)
return ec2MarketplaceListingResult{}, h.recordDeadListing(ctx, purchaseID, claim, result)
}

// Persist the listing ID and state. On DB failure, attempt a compensating
Expand All @@ -226,17 +225,17 @@ func (h *Handler) createAndPersistListing(ctx context.Context, purchaseID string
if isListingEndedConflict(dbErr) {
// The cancel endpoint found this attempt's listing through its
// token, canceled it on AWS and recorded that first.
return ec2svc.MarketplaceListingResult{}, listingCanceledConcurrently(result.ListingID)
return ec2MarketplaceListingResult{}, listingCanceledConcurrently(result.ListingID)
}
logging.Errorf("marketplace: listing created (%s / %s) but DB update failed: %v; attempting rollback", result.ListingID, result.State, dbErr)
compCtx, cancelComp := context.WithTimeout(context.WithoutCancel(ctx), marketplaceCompensationTimeout)
defer cancelComp()
if _, rollbackErr := ec2Client.CancelMarketplaceListing(compCtx, result.ListingID); rollbackErr != nil {
return ec2svc.MarketplaceListingResult{}, h.keepUncanceledListing(compCtx, purchaseID, result, rollbackErr)
return ec2MarketplaceListingResult{}, h.keepUncanceledListing(compCtx, purchaseID, result, rollbackErr)
}
logging.Warnf("marketplace: listing %s rolled back (canceled) after DB failure", result.ListingID)
h.recordRolledBackListing(compCtx, purchaseID, claim, result.ListingID)
return ec2svc.MarketplaceListingResult{}, fmt.Errorf("listing created but could not be persisted; listing has been rolled back: %w", dbErr)
return ec2MarketplaceListingResult{}, fmt.Errorf("listing created but could not be persisted; listing has been rolled back: %w", dbErr)
}

return result, nil
Expand All @@ -254,7 +253,7 @@ func isDeadListingState(state string) bool {
// recordDeadListing handles a listing of this attempt that is canceled or
// closed on AWS: it records the listing, which also drops the attempt's token,
// so the next attempt starts fresh instead of AWS replaying the dead listing.
func (h *Handler) recordDeadListing(ctx context.Context, purchaseID string, claim *config.MarketplaceListingClaim, listing ec2svc.MarketplaceListingResult) error {
func (h *Handler) recordDeadListing(ctx context.Context, purchaseID string, claim *config.MarketplaceListingClaim, listing ec2MarketplaceListingResult) error {
if err := h.config.UpdatePurchaseHistoryListing(ctx, purchaseID, listing.ListingID, listing.State); err != nil {
logging.Errorf("marketplace: failed to record dead listing %s for purchase %s: %v", listing.ListingID, purchaseID, err)
h.releaseMarketplaceClaim(ctx, purchaseID, claim, false)
Expand Down
9 changes: 4 additions & 5 deletions internal/api/handler_marketplace_cancel.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@ import (
"strings"

"github.com/LeanerCloud/cloud-commitments-go/pkg/logging"
ec2svc "github.com/LeanerCloud/cloud-commitments-go/providers/aws/services/ec2"
"github.com/LeanerCloud/cloud-commitments-platform/internal/auth"
"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
"github.com/aws/aws-lambda-go/events"
Expand Down Expand Up @@ -178,16 +177,16 @@ func (h *Handler) cancelListing(ctx context.Context, purchaseID string, seen *co
// endedMarketplaceListing reports whether listingID has already ended on AWS,
// read after its cancel failed. A role-assumption failure would fail the read
// the same way, so it is not retried.
func endedMarketplaceListing(ctx context.Context, ec2Client marketplaceEC2Client, listingID string, cancelErr error) (ec2svc.MarketplaceListingResult, bool) {
func endedMarketplaceListing(ctx context.Context, ec2Client marketplaceEC2Client, listingID string, cancelErr error) (ec2MarketplaceListingResult, bool) {
if isRoleAssumptionFailure(cancelErr) {
return ec2svc.MarketplaceListingResult{}, false
return ec2MarketplaceListingResult{}, false
}
listing, err := ec2Client.DescribeMarketplaceListing(ctx, listingID)
if err != nil || !isDeadListingState(listing.State) {
return ec2svc.MarketplaceListingResult{}, false
return ec2MarketplaceListingResult{}, false
}
logging.Warnf("marketplace: listing %s is already %s on AWS (cancel said: %v); recording it", listingID, listing.State, cancelErr)
return ec2svc.MarketplaceListingResult{ListingID: listingID, State: listing.State}, true
return ec2MarketplaceListingResult{ListingID: listingID, State: listing.State}, true
}

// resolveListingAttempt records listingID in state on a pending row, only
Expand Down
Loading
Loading